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

✨(documents) allow retrieving versions (list and detail)

Browse Source 
Versions are retrieved directly from object storage and served on API
endpoints. We make sure a user who is given access to a document will
only see versions that were created after s.he gained access.
This commit is contained in:
Samuel Paccoud - DINUM
2024-04-08 23:37:15 +02:00
committed by Anthony LC
parent 8e262da8f5
commit 130e7a8c99
7 changed files with 765 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
src/backend/core/api/permissions.py
View File

@@ -3,6 +3,10 @@ from django.core import exceptions
from rest_framework import permissions
ACTION_FOR_METHOD_TO_PERMISSION = {
"versions_detail": {"DELETE": "versions_destroy", "GET": "versions_retrieve"}
}
class IsAuthenticated(permissions.BasePermission):
"""
@@ -60,4 +64,9 @@ class AccessPermission(permissions.BasePermission):
def has_object_permission(self, request, view, obj):
"""Check permission for a given object."""
abilities = obj.get_abilities(request.user)
return abilities.get(view.action, False)
action = view.action
try:
action = ACTION_FOR_METHOD_TO_PERMISSION[view.action][request.method]
except KeyError:
pass
return abilities.get(action, False)

60
src/backend/core/api/viewsets.py
View File

@@ -1,4 +1,5 @@
"""API endpoints"""
import json
from io import BytesIO
from django.contrib.postgres.aggregates import ArrayAgg
@@ -7,8 +8,9 @@ from django.db.models import (
Q,
Subquery,
)
from django.http import FileResponse
from django.http import FileResponse, Http404
from botocore.exceptions import ClientError
from rest_framework import (
decorators,
exceptions,
@@ -291,6 +293,62 @@ class DocumentViewSet(
resource_field_name = "document"
queryset = models.Document.objects.all()
@decorators.action(detail=True, methods=["get"], url_path="versions")
def versions_list(self, request, *args, **kwargs):
"""
Return the document's versions but only those created after the user got access
to the document
"""
document = self.get_object()
from_datetime = min(
access.created_at
for access in document.accesses.filter(
Q(user=request.user) | Q(team__in=request.user.get_teams()),
)
)
return drf_response.Response(
document.get_versions_slice(from_datetime=from_datetime)
)
@decorators.action(
detail=True,
methods=["get", "delete"],
url_path="versions/(?P<version_id>[0-9a-f-]{36})",
)
# pylint: disable=unused-argument
def versions_detail(self, request, pk, version_id, *args, **kwargs):
"""Custom action to retrieve a specific version of a document"""
document = self.get_object()
try:
response = document.get_content_response(version_id=version_id)
except (FileNotFoundError, ClientError) as err:
raise Http404 from err
# Don't let users access versions that were created before they were given access
# to the document
from_datetime = min(
access.created_at
for access in document.accesses.filter(
Q(user=request.user) | Q(team__in=request.user.get_teams()),
)
)
if response["LastModified"] < from_datetime:
raise Http404
if request.method == "DELETE":
response = document.delete_version(version_id)
return drf_response.Response(
status=response["ResponseMetadata"]["HTTPStatusCode"]
)
return drf_response.Response(
{
"content": json.loads(response["Body"].read()),
"last_modified": response["LastModified"],
}
)
class DocumentAccessViewSet(
ResourceAccessViewsetMixin,

103
src/backend/core/models.py
View File

@@ -21,6 +21,7 @@ from django.utils.translation import gettext_lazy as _
import frontmatter
import markdown
from botocore.exceptions import ClientError
from timezone_field import TimeZoneField
from weasyprint import CSS, HTML
from weasyprint.text.fonts import FontConfiguration
@@ -264,16 +265,23 @@ class Document(BaseModel):
def __str__(self):
return self.title
@property
def file_key(self):
"""Key of the object storage file to which the document content is stored"""
if not self.pk:
return None
return str(self.pk)
@property
def content(self):
"""Return the json content from object storage if available"""
if self._content is None and self.id:
try:
# Load content from object storage
with default_storage.open(str(self.id)) as f:
self._content = json.load(f)
except FileNotFoundError:
response = self.get_content_response()
except (FileNotFoundError, ClientError):
pass
else:
self._content = json.loads(response["Body"].read())
return self._content
@content.setter
@@ -285,12 +293,18 @@ class Document(BaseModel):
raise ValueError("content should be a json object.")
self._content = content
def get_content_response(self, version_id=""):
"""Get the content in a specific version of the document"""
return default_storage.connection.meta.client.get_object(
Bucket=default_storage.bucket_name, Key=self.file_key, VersionId=version_id
)
def save(self, *args, **kwargs):
"""Write content to object storage only if _content has changed."""
super().save(*args, **kwargs)
if self._content:
file_key = str(self.pk)
file_key = self.file_key
bytes_content = json.dumps(self._content).encode("utf-8")
if default_storage.exists(file_key):
@@ -307,6 +321,81 @@ class Document(BaseModel):
content_file = ContentFile(bytes_content)
default_storage.save(file_key, content_file)
def get_versions_slice(
self, from_version_id="", from_datetime=None, page_size=None
):
"""Get document versions from object storage with pagination and starting conditions"""
# /!\ Trick here /!\
# The "KeyMarker" and "VersionIdMarker" fields must either be both set or both not set.
# The error we get otherwise is not helpful at all.
token = {}
if from_version_id:
token.update(
{"KeyMarker": self.file_key, "VersionIdMarker": from_version_id}
)
if from_datetime:
response = default_storage.connection.meta.client.list_object_versions(
Bucket=default_storage.bucket_name,
Prefix=self.file_key,
MaxKeys=settings.S3_VERSIONS_PAGE_SIZE,
**token,
)
# Find the first version after the given datetime
version = None
for version in response.get("Versions", []):
if version["LastModified"] >= from_datetime:
token = {
"KeyMarker": self.file_key,
"VersionIdMarker": version["VersionId"],
}
break
else:
if version is None or version["LastModified"] < from_datetime:
if response["NextVersionIdMarker"]:
return self.get_versions_slice(
from_version_id=response["NextVersionIdMarker"],
page_size=settings.S3_VERSIONS_PAGE_SIZE,
from_datetime=from_datetime,
)
return {
"next_version_id_marker": "",
"is_truncated": False,
"versions": [],
}
response = default_storage.connection.meta.client.list_object_versions(
Bucket=default_storage.bucket_name,
Prefix=self.file_key,
MaxKeys=min(page_size, settings.S3_VERSIONS_PAGE_SIZE)
if page_size
else settings.S3_VERSIONS_PAGE_SIZE,
**token,
)
return {
"next_version_id_marker": response["NextVersionIdMarker"],
"is_truncated": response["IsTruncated"],
"versions": [
{
key_snake: version[key_camel]
for key_camel, key_snake in [
("ETag", "etag"),
("IsLatest", "is_latest"),
("LastModified", "last_modified"),
("VersionId", "version_id"),
]
}
for version in response.get("Versions", [])
],
}
def delete_version(self, version_id):
"""Delete a version from object storage given its version id"""
return default_storage.connection.meta.client.delete_object(
Bucket=default_storage.bucket_name, Key=self.file_key, VersionId=version_id
)
def get_abilities(self, user):
"""
Compute and return abilities for a given user on the document.
@@ -316,9 +405,13 @@ class Document(BaseModel):
set(roles).intersection({RoleChoices.OWNER, RoleChoices.ADMIN})
)
can_get = self.is_public or bool(roles)
can_get_versions = bool(roles)
return {
"destroy": RoleChoices.OWNER in roles,
"versions_destroy": is_owner_or_admin,
"versions_list": can_get_versions,
"versions_retrieve": can_get_versions,
"manage_accesses": is_owner_or_admin,
"update": is_owner_or_admin,
"partial_update": is_owner_or_admin,

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

@@ -24,6 +24,9 @@ def test_api_documents_retrieve_anonymous_public():
"partial_update": False,
"retrieve": True,
"update": False,
"versions_destroy": False,
"versions_list": False,
"versions_retrieve": False,
},
"accesses": [],
"title": document.title,
@@ -66,6 +69,9 @@ def test_api_documents_retrieve_authenticated_unrelated_public():
"partial_update": False,
"retrieve": True,
"update": False,
"versions_destroy": False,
"versions_list": False,
"versions_retrieve": False,
},
"accesses": [],
"title": document.title,

526
src/backend/core/tests/test_api_document_versions.py Normal file
View File

@@ -0,0 +1,526 @@
"""
Test document versions API endpoints for users in impress's core app.
"""
import random
import time
import pytest
from rest_framework.test import APIClient
from core import factories, models
from core.tests.conftest import TEAM, USER, VIA
pytestmark = pytest.mark.django_db
def test_api_document_versions_list_anonymous_public():
"""
Anonymous users should not be allowed to list document versions for a public document.
"""
document = factories.DocumentFactory(is_public=True)
factories.UserDocumentAccessFactory.create_batch(2, document=document)
response = APIClient().get(f"/api/v1.0/documents/{document.id!s}/versions/")
assert response.status_code == 401
assert response.json() == {
"detail": "Authentication credentials were not provided."
}
def test_api_document_versions_list_anonymous_private():
"""
Anonymous users should not be allowed to find document versions for a private document.
"""
document = factories.DocumentFactory(is_public=False)
factories.UserDocumentAccessFactory.create_batch(2, document=document)
response = APIClient().get(f"/api/v1.0/documents/{document.id!s}/versions/")
assert response.status_code == 404
assert response.json() == {"detail": "Not found."}
def test_api_document_versions_list_authenticated_unrelated_public():
"""
Authenticated users should not be allowed to list document versions for a public document
to which they are not related.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
document = factories.DocumentFactory(is_public=True)
factories.UserDocumentAccessFactory.create_batch(3, document=document)
# The versions of another document to which the user is related should not be listed either
factories.UserDocumentAccessFactory(user=user)
response = client.get(
f"/api/v1.0/documents/{document.id!s}/versions/",
)
assert response.status_code == 403
assert response.json() == {
"detail": "You do not have permission to perform this action."
}
def test_api_document_versions_list_authenticated_unrelated_private():
"""
Authenticated users should not be allowed to find document versions for a private document
to which they are not related.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
document = factories.DocumentFactory(is_public=False)
factories.UserDocumentAccessFactory.create_batch(3, document=document)
# The versions of another document to which the user is related should not be listed either
factories.UserDocumentAccessFactory(user=user)
response = client.get(
f"/api/v1.0/documents/{document.id!s}/versions/",
)
assert response.status_code == 404
assert response.json() == {"detail": "Not found."}
@pytest.mark.parametrize("via", VIA)
def test_api_document_versions_list_authenticated_related(via, mock_user_get_teams):
"""
Authenticated users should be able to list document versions for a document
to which they are directly related, whatever their role in the document.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
document = factories.DocumentFactory()
if via == USER:
models.DocumentAccess.objects.create(
document=document,
user=user,
role=random.choice(models.RoleChoices.choices)[0],
)
elif via == TEAM:
mock_user_get_teams.return_value = ["lasuite", "unknown"]
models.DocumentAccess.objects.create(
document=document,
team="lasuite",
role=random.choice(models.RoleChoices.choices)[0],
)
# Other versions of documents to which the user has access should not be listed
factories.UserDocumentAccessFactory(user=user)
# A version created before the user got access should be hidden
response = client.get(
f"/api/v1.0/documents/{document.id!s}/versions/",
)
assert response.status_code == 200
content = response.json()
assert len(content["versions"]) == 0
# Add a new version to the document
document.content = {"foo": "bar"}
document.save()
response = client.get(
f"/api/v1.0/documents/{document.id!s}/versions/",
)
assert response.status_code == 200
content = response.json()
assert len(content["versions"]) == 1
assert content["next_version_id_marker"] == ""
assert content["is_truncated"] is False
def test_api_document_versions_retrieve_anonymous_public():
"""
Anonymous users should not be allowed to retrieve specific versions for a public document.
"""
document = factories.DocumentFactory(is_public=True)
version_id = document.get_versions_slice()["versions"][0]["version_id"]
url = f"/api/v1.0/documents/{document.id!s}/versions/{version_id:s}/"
response = APIClient().get(url)
assert response.status_code == 401
assert response.json() == {
"detail": "Authentication credentials were not provided."
}
def test_api_document_versions_retrieve_anonymous_private():
"""
Anonymous users should not be allowed to find specific versions for a private document.
"""
document = factories.DocumentFactory(is_public=False)
version_id = document.get_versions_slice()["versions"][0]["version_id"]
url = f"/api/v1.0/documents/{document.id!s}/versions/{version_id:s}/"
response = APIClient().get(url)
assert response.status_code == 404
assert response.json() == {"detail": "Not found."}
def test_api_document_versions_retrieve_authenticated_unrelated_public():
"""
Authenticated users should not be allowed to retrieve specific versions for a public
document to which they are not related.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
document = factories.DocumentFactory(is_public=True)
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 == 403
assert response.json() == {
"detail": "You do not have permission to perform this action."
}
def test_api_document_versions_retrieve_authenticated_unrelated_private():
"""
Authenticated users should not be allowed to find specific versions for a private document
to which they are not related.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
document = factories.DocumentFactory(is_public=False)
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
assert response.json() == {"detail": "Not found."}
@pytest.mark.parametrize("via", VIA)
def test_api_document_versions_retrieve_authenticated_related(via, mock_user_get_teams):
"""
A user who is related to a document should be allowed to retrieve the
associated document user accesses.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
document = factories.DocumentFactory()
version_id = document.get_versions_slice()["versions"][0]["version_id"]
if via == USER:
factories.UserDocumentAccessFactory(document=document, user=user)
elif via == TEAM:
mock_user_get_teams.return_value = ["lasuite", "unknown"]
factories.TeamDocumentAccessFactory(document=document, team="lasuite")
# Versions created before the document was shared should not be available to 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 make it available to the user
time.sleep(1) # minio stores datetimes with the precision of a second
document.content = {"foo": "bar"}
document.save()
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"] == {"foo": "bar"}
def test_api_document_versions_create_anonymous():
"""Anonymous users should not be allowed to create document versions."""
document = factories.DocumentFactory()
response = APIClient().post(
f"/api/v1.0/documents/{document.id!s}/versions/",
{"foo": "bar"},
format="json",
)
assert response.status_code == 401
assert response.json() == {
"detail": "Authentication credentials were not provided."
}
def test_api_document_versions_create_authenticated_unrelated():
"""
Authenticated users should not be allowed to create document versions for a document to
which they are not related.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
document = factories.DocumentFactory()
response = client.post(
f"/api/v1.0/documents/{document.id!s}/versions/",
{"foo": "bar"},
format="json",
)
assert response.status_code == 405
@pytest.mark.parametrize("via", VIA)
def test_api_document_versions_create_authenticated_related(via, mock_user_get_teams):
"""
Authenticated users related to a document should not be allowed to create document versions
whatever their role.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
document = factories.DocumentFactory()
if via == USER:
factories.UserDocumentAccessFactory(document=document, user=user)
elif via == TEAM:
mock_user_get_teams.return_value = ["lasuite", "unknown"]
factories.TeamDocumentAccessFactory(document=document, team="lasuite")
response = client.post(
f"/api/v1.0/documents/{document.id!s}/versions/",
{"foo": "bar"},
format="json",
)
assert response.status_code == 405
def test_api_document_versions_update_anonymous():
"""Anonymous users should not be allowed to update a document version."""
access = factories.UserDocumentAccessFactory()
version_id = access.document.get_versions_slice()["versions"][0]["version_id"]
response = APIClient().put(
f"/api/v1.0/documents/{access.document_id!s}/versions/{version_id:s}/",
{"foo": "bar"},
format="json",
)
assert response.status_code == 401
def test_api_document_versions_update_authenticated_unrelated():
"""
Authenticated users should not be allowed to update a document version for a document to which
they are not related.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
access = factories.UserDocumentAccessFactory()
version_id = access.document.get_versions_slice()["versions"][0]["version_id"]
response = client.put(
f"/api/v1.0/documents/{access.document_id!s}/versions/{version_id:s}/",
{"foo": "bar"},
format="json",
)
assert response.status_code == 405
@pytest.mark.parametrize("via", VIA)
def test_api_document_versions_update_authenticated_related(via, mock_user_get_teams):
"""
Authenticated users with access to a document should not be able to update its versions
whatever their role.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
document = factories.DocumentFactory()
version_id = document.get_versions_slice()["versions"][0]["version_id"]
if via == USER:
factories.UserDocumentAccessFactory(document=document, user=user)
elif via == TEAM:
mock_user_get_teams.return_value = ["lasuite", "unknown"]
factories.TeamDocumentAccessFactory(document=document, team="lasuite")
response = client.put(
f"/api/v1.0/documents/{document.id!s}/versions/{version_id!s}/",
{"foo": "bar"},
format="json",
)
assert response.status_code == 405
# Delete
def test_api_document_versions_delete_anonymous():
"""Anonymous users should not be allowed to destroy a document version."""
access = factories.UserDocumentAccessFactory()
response = APIClient().delete(
f"/api/v1.0/documents/{access.document_id!s}/versions/{access.id!s}/",
)
assert response.status_code == 401
def test_api_document_versions_delete_authenticated_public():
"""
Authenticated users should not be allowed to delete a document version for a
public document to which they are not related.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
document = factories.DocumentFactory(is_public=True)
version_id = document.get_versions_slice()["versions"][0]["version_id"]
response = client.delete(
f"/api/v1.0/documents/{document.id!s}/versions/{version_id:s}/",
)
assert response.status_code == 403
def test_api_document_versions_delete_authenticated_private():
"""
Authenticated users should not be allowed to find a document version to delete it
for a private document to which they are not related.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
document = factories.DocumentFactory(is_public=False)
version_id = document.get_versions_slice()["versions"][0]["version_id"]
response = client.delete(
f"/api/v1.0/documents/{document.id!s}/versions/{version_id:s}/",
)
assert response.status_code == 404
assert response.json() == {"detail": "Not found."}
@pytest.mark.parametrize("via", VIA)
def test_api_document_versions_delete_member(via, mock_user_get_teams):
"""
Authenticated users should not be allowed to delete a document version for a
document in which they are a simple member.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
document = factories.DocumentFactory()
if via == USER:
factories.UserDocumentAccessFactory(document=document, user=user, role="member")
elif via == TEAM:
mock_user_get_teams.return_value = ["lasuite", "unknown"]
factories.TeamDocumentAccessFactory(
document=document, team="lasuite", role="member"
)
# Create a new version should make it available to the user
time.sleep(1) # minio stores datetimes with the precision of a second
document.content = {"foo": "bar"}
document.save()
versions = document.get_versions_slice()["versions"]
assert len(versions) == 2
version_id = versions[1]["version_id"]
response = client.delete(
f"/api/v1.0/documents/{document.id!s}/versions/{version_id:s}/",
)
assert response.status_code == 403
version_id = versions[0]["version_id"]
response = client.delete(
f"/api/v1.0/documents/{document.id!s}/versions/{version_id:s}/",
)
assert response.status_code == 403
versions = document.get_versions_slice()["versions"]
assert len(versions) == 2
@pytest.mark.parametrize("via", VIA)
def test_api_document_versions_delete_administrator_or_owner(via, mock_user_get_teams):
"""
Users who are administrator or owner of a document should be allowed to delete a version.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
document = factories.DocumentFactory()
role = random.choice(["administrator", "owner"])
if via == USER:
factories.UserDocumentAccessFactory(document=document, user=user, role=role)
elif via == TEAM:
mock_user_get_teams.return_value = ["lasuite", "unknown"]
factories.TeamDocumentAccessFactory(
document=document, team="lasuite", role=role
)
# Create a new version should make it available to the user
time.sleep(1) # minio stores datetimes with the precision of a second
document.content = {"foo": "bar"}
document.save()
versions = document.get_versions_slice()["versions"]
assert len(versions) == 2
version_id = versions[1]["version_id"]
response = client.delete(
f"/api/v1.0/documents/{document.id!s}/versions/{version_id:s}/",
)
# 404 because the version was created before the user was given access to the document
assert response.status_code == 404
version_id = versions[0]["version_id"]
response = client.delete(
f"/api/v1.0/documents/{document.id!s}/versions/{version_id:s}/",
)
assert response.status_code == 204
versions = document.get_versions_slice()["versions"]
assert len(versions) == 1

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

@@ -6,7 +6,6 @@ from django.core.exceptions import ValidationError
from django.core.files.storage import default_storage
import pytest
import requests
from core import factories, models
@@ -61,6 +60,9 @@ def test_models_documents_get_abilities_anonymous_public():
"update": False,
"manage_accesses": False,
"partial_update": False,
"versions_destroy": False,
"versions_list": False,
"versions_retrieve": False,
}
@@ -74,10 +76,13 @@ def test_models_documents_get_abilities_anonymous_not_public():
"update": False,
"manage_accesses": False,
"partial_update": False,
"versions_destroy": False,
"versions_list": False,
"versions_retrieve": False,
}
def test_models_documents_get_abilities_authenticated_public():
def test_models_documents_get_abilities_authenticated_unrelated_public():
"""Check abilities returned for an authenticated user if the user is public."""
document = factories.DocumentFactory(is_public=True)
abilities = document.get_abilities(factories.UserFactory())
@@ -87,10 +92,13 @@ def test_models_documents_get_abilities_authenticated_public():
"update": False,
"manage_accesses": False,
"partial_update": False,
"versions_destroy": False,
"versions_list": False,
"versions_retrieve": False,
}
def test_models_documents_get_abilities_authenticated_not_public():
def test_models_documents_get_abilities_authenticated_unrelated_not_public():
"""Check abilities returned for an authenticated user if the document is private."""
document = factories.DocumentFactory(is_public=False)
abilities = document.get_abilities(factories.UserFactory())
@@ -100,6 +108,9 @@ def test_models_documents_get_abilities_authenticated_not_public():
"update": False,
"manage_accesses": False,
"partial_update": False,
"versions_destroy": False,
"versions_list": False,
"versions_retrieve": False,
}
@@ -114,6 +125,9 @@ def test_models_documents_get_abilities_owner():
"update": True,
"manage_accesses": True,
"partial_update": True,
"versions_destroy": True,
"versions_list": True,
"versions_retrieve": True,
}
@@ -127,6 +141,9 @@ def test_models_documents_get_abilities_administrator():
"update": True,
"manage_accesses": True,
"partial_update": True,
"versions_destroy": True,
"versions_list": True,
"versions_retrieve": True,
}
@@ -143,6 +160,9 @@ def test_models_documents_get_abilities_member_user(django_assert_num_queries):
"update": False,
"manage_accesses": False,
"partial_update": False,
"versions_destroy": False,
"versions_list": True,
"versions_retrieve": True,
}
@@ -160,23 +180,51 @@ def test_models_documents_get_abilities_preset_role(django_assert_num_queries):
"update": False,
"manage_accesses": False,
"partial_update": False,
"versions_destroy": False,
"versions_list": True,
"versions_retrieve": True,
}
def test_models_documents_file_upload_to_minio():
"""Validate read/write from/to minio"""
document = factories.DocumentFactory()
document.content = {"foé": "çar"}
document.save()
def test_models_documents_get_versions_slice(settings):
"""
The "get_versions_slice" method should allow navigating all versions of
the document with pagination.
"""
settings.S3_VERSIONS_PAGE_SIZE = 4
# Check that the file exists in MinIO:
file_key = str(document.pk)
# - through the storage backend
assert default_storage.exists(file_key) is True
# - directly from minio
signed_url = default_storage.url(file_key)
response = requests.get(signed_url, timeout=1)
assert response.json() == {"foé": "çar"}
# Create a document with 7 versions
document = factories.DocumentFactory()
for i in range(6):
document.content = {"foo": f"bar{i:d}"}
document.save()
# Add a version not related to the first document
factories.DocumentFactory()
# - Get default max versions
response = document.get_versions_slice()
assert response["is_truncated"] is True
assert len(response["versions"]) == 4
assert response["next_version_id_marker"] != ""
expected_keys = ["etag", "is_latest", "last_modified", "version_id"]
for i in range(4):
assert list(response["versions"][i].keys()) == expected_keys
# - Get page 2
response = document.get_versions_slice(
from_version_id=response["next_version_id_marker"]
)
assert response["is_truncated"] is False
assert len(response["versions"]) == 3
assert response["next_version_id_marker"] == ""
# - Get custom max versions
response = document.get_versions_slice(page_size=2)
assert response["is_truncated"] is True
assert len(response["versions"]) == 2
assert response["next_version_id_marker"] != ""
def test_models_documents_version_duplicate():

2
src/backend/impress/settings.py
View File

@@ -137,6 +137,8 @@ class Base(Configuration):
environ_prefix=None,
)
S3_VERSIONS_PAGE_SIZE = 50
# Internationalization
# https://docs.djangoproject.com/en/3.1/topics/i18n/