✨(backend) add duplicate action to the document API endpoint

We took this opportunity to refactor the way access is controlled on
media attachments. We now add the media key to a list on the document
instance each time a media is uploaded to a document. This list is
passed along when a document is duplicated, allowing us to grant
access to readers on the new document, even if they don't have or
lost access to the original document.

We also propose an option to reproduce the same access rights on the
duplicate document as what was in place on the original document.
This can be requested by passing the "with_accesses=true" option in
the query string.

The tricky point is that we need to extract attachment keys from the
existing documents and set them on the new "attachments" field that is
now used to track access rights on media files.
This commit is contained in:
Samuel Paccoud - DINUM
2025-03-24 10:43:45 +01:00
committed by Manuel Raynaud
parent 6976bb7c78
commit 34a208a80d
19 changed files with 1066 additions and 79 deletions
@@ -67,10 +67,12 @@ def test_api_documents_attachment_upload_anonymous_success():
file_path = response.json()["file"]
match = pattern.search(file_path)
file_id = match.group(1)
# Validate that file_id is a valid UUID
uuid.UUID(file_id)
document.refresh_from_db()
assert document.attachments == [f"{document.id!s}/attachments/{file_id!s}.png"]
# Now, check the metadata of the uploaded file
key = file_path.replace("/media", "")
file_head = default_storage.connection.meta.client.head_object(
@@ -112,6 +114,9 @@ def test_api_documents_attachment_upload_authenticated_forbidden(reach, role):
"detail": "You do not have permission to perform this action."
}
document.refresh_from_db()
assert document.attachments == []
@pytest.mark.parametrize(
"reach, role",
@@ -122,8 +127,8 @@ def test_api_documents_attachment_upload_authenticated_forbidden(reach, role):
)
def test_api_documents_attachment_upload_authenticated_success(reach, role):
"""
Autenticated who are not related to a document should be able to upload a file
if the link reach and role permit it.
Autenticated users who are not related to a document should be able to upload
a file when the link reach and role permit it.
"""
user = factories.UserFactory()
@@ -145,6 +150,9 @@ def test_api_documents_attachment_upload_authenticated_success(reach, role):
# Validate that file_id is a valid UUID
uuid.UUID(file_id)
document.refresh_from_db()
assert document.attachments == [f"{document.id!s}/attachments/{file_id!s}.png"]
@pytest.mark.parametrize("via", VIA)
def test_api_documents_attachment_upload_reader(via, mock_user_teams):
@@ -175,6 +183,9 @@ def test_api_documents_attachment_upload_reader(via, mock_user_teams):
"detail": "You do not have permission to perform this action."
}
document.refresh_from_db()
assert document.attachments == []
@pytest.mark.parametrize("role", ["editor", "administrator", "owner"])
@pytest.mark.parametrize("via", VIA)
@@ -211,6 +222,9 @@ def test_api_documents_attachment_upload_success(via, role, mock_user_teams):
# Validate that file_id is a valid UUID
uuid.UUID(file_id)
document.refresh_from_db()
assert document.attachments == [f"{document.id!s}/attachments/{file_id!s}.png"]
# Now, check the metadata of the uploaded file
key = file_path.replace("/media", "")
file_head = default_storage.connection.meta.client.head_object(
@@ -236,6 +250,9 @@ def test_api_documents_attachment_upload_invalid(client):
assert response.status_code == 400
assert response.json() == {"file": ["No file was submitted."]}
document.refresh_from_db()
assert document.attachments == []
def test_api_documents_attachment_upload_size_limit_exceeded(settings):
"""The uploaded file should not exceeed the maximum size in settings."""
@@ -258,6 +275,9 @@ def test_api_documents_attachment_upload_size_limit_exceeded(settings):
assert response.status_code == 400
assert response.json() == {"file": ["File size exceeds the maximum limit of 1 MB."]}
document.refresh_from_db()
assert document.attachments == []
@pytest.mark.parametrize(
"name,content,extension,content_type",
@@ -293,6 +313,11 @@ def test_api_documents_attachment_upload_fix_extension(
match = pattern.search(file_path)
file_id = match.group(1)
document.refresh_from_db()
assert document.attachments == [
f"{document.id!s}/attachments/{file_id!s}.{extension:s}"
]
assert "-unsafe" in file_id
# Validate that file_id is a valid UUID
file_id = file_id.replace("-unsafe", "")
@@ -323,6 +348,9 @@ def test_api_documents_attachment_upload_empty_file():
assert response.status_code == 400
assert response.json() == {"file": ["The submitted file is empty."]}
document.refresh_from_db()
assert document.attachments == []
def test_api_documents_attachment_upload_unsafe():
"""A file with an unsafe mime type should be tagged as such."""
@@ -345,6 +373,9 @@ def test_api_documents_attachment_upload_unsafe():
match = pattern.search(file_path)
file_id = match.group(1)
document.refresh_from_db()
assert document.attachments == [f"{document.id!s}/attachments/{file_id!s}.exe"]
assert "-unsafe" in file_id
# Validate that file_id is a valid UUID
file_id = file_id.replace("-unsafe", "")
@@ -0,0 +1,206 @@
"""
Test file uploads API endpoint for users in impress's core app.
"""
import base64
import uuid
from io import BytesIO
from urllib.parse import urlparse
from django.conf import settings
from django.core.files.storage import default_storage
from django.utils import timezone
import pytest
import requests
import y_py
from rest_framework.test import APIClient
from core import factories, models
pytestmark = pytest.mark.django_db
PIXEL = (
b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06\x00"
b"\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\xf8\xff\xff?\x00\x05\xfe\x02\xfe"
b"\xa7V\xbd\xfa\x00\x00\x00\x00IEND\xaeB`\x82"
)
def get_image_refs(document_id):
"""Generate an image key for testing."""
image_key = f"{document_id!s}/attachments/{uuid.uuid4()!s}.png"
default_storage.connection.meta.client.put_object(
Bucket=default_storage.bucket_name,
Key=image_key,
Body=BytesIO(PIXEL),
ContentType="image/png",
)
return image_key, f"http://localhost/media/{image_key:s}"
def test_api_documents_duplicate_forbidden():
"""A user who doesn't have read access to a document should not be allowed to duplicate it."""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
document = factories.DocumentFactory(
link_reach="restricted",
users=[factories.UserFactory()],
title="my document",
)
response = client.post(f"/api/v1.0/documents/{document.id!s}/duplicate/")
assert response.status_code == 403
assert models.Document.objects.count() == 1
def test_api_documents_duplicate_anonymous():
"""Anonymous users should not be able to duplicate documents even with read access."""
document = factories.DocumentFactory(link_reach="public")
response = APIClient().post(f"/api/v1.0/documents/{document.id!s}/duplicate/")
assert response.status_code == 401
assert models.Document.objects.count() == 1
@pytest.mark.parametrize("index", range(3))
def test_api_documents_duplicate_success(index):
"""
Anonymous users should be able to retrieve attachments linked to a public document.
Accesses should not be duplicated if the user does not request it specifically.
Attachments that are not in the content should not be passed for access in the
duplicated document's "attachments" list.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
document_ids = [uuid.uuid4() for _ in range(3)]
image_refs = [get_image_refs(doc_id) for doc_id in document_ids]
# Create document content with the first image only
ydoc = y_py.YDoc() # pylint: disable=no-member
with ydoc.begin_transaction() as txn:
xml_fragment = ydoc.get_xml_element("document-store")
xml_fragment.push_xml_element(txn, "image").set_attribute(
txn, "src", image_refs[0][1]
)
update = y_py.encode_state_as_update(ydoc) # pylint: disable=no-member
base64_content = base64.b64encode(update).decode("utf-8")
# Create documents
document = factories.DocumentFactory(
id=document_ids[index],
content=base64_content,
link_reach="restricted",
users=[user, factories.UserFactory()],
title="document with an image",
attachments=[key for key, _ in image_refs],
)
factories.DocumentFactory(id=document_ids[(index + 1) % 3])
# Don't create document for third ID to check that it doesn't impact access to attachments
# Duplicate the document via the API endpoint
response = client.post(f"/api/v1.0/documents/{document.id}/duplicate/")
assert response.status_code == 201
duplicated_document = models.Document.objects.get(id=response.json()["id"])
assert duplicated_document.title == "Copy of document with an image"
assert duplicated_document.content == document.content
assert duplicated_document.creator == user
assert duplicated_document.link_reach == "restricted"
assert duplicated_document.link_role == "reader"
assert duplicated_document.duplicated_from == document
assert duplicated_document.attachments == [
image_refs[0][0]
] # Only the first image key
assert duplicated_document.get_parent() == document.get_parent()
assert duplicated_document.path == document.get_next_sibling().path
# Check that accesses were not duplicated.
# The user who did the duplicate is forced as owner
assert duplicated_document.accesses.count() == 1
access = duplicated_document.accesses.first()
assert access.user == user
assert access.role == "owner"
# Ensure access persists after the owner loses access to the original document
models.DocumentAccess.objects.filter(document=document).delete()
response = client.get(
"/api/v1.0/documents/media-auth/", HTTP_X_ORIGINAL_URL=image_refs[0][1]
)
assert response.status_code == 200
authorization = response["Authorization"]
assert "AWS4-HMAC-SHA256 Credential=" in authorization
assert (
"SignedHeaders=host;x-amz-content-sha256;x-amz-date, Signature="
in authorization
)
assert response["X-Amz-Date"] == timezone.now().strftime("%Y%m%dT%H%M%SZ")
s3_url = urlparse(settings.AWS_S3_ENDPOINT_URL)
response = requests.get(
f"{settings.AWS_S3_ENDPOINT_URL:s}/impress-media-storage/{image_refs[0][0]:s}",
headers={
"authorization": authorization,
"x-amz-date": response["x-amz-date"],
"x-amz-content-sha256": response["x-amz-content-sha256"],
"Host": f"{s3_url.hostname:s}:{s3_url.port:d}",
},
timeout=1,
)
assert response.content == PIXEL
# Ensure the other images are not accessible
for _, url in image_refs[1:]:
response = client.get(
"/api/v1.0/documents/media-auth/", HTTP_X_ORIGINAL_URL=url
)
assert response.status_code == 403
def test_api_documents_duplicate_with_accesses():
"""Accesses should be duplicated if the user requests it specifically."""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
document = factories.DocumentFactory(
users=[user],
title="document with accesses",
)
user_access = factories.UserDocumentAccessFactory(document=document)
team_access = factories.TeamDocumentAccessFactory(document=document)
# Duplicate the document via the API endpoint requesting to duplicate accesses
response = client.post(
f"/api/v1.0/documents/{document.id!s}/duplicate/",
{"with_accesses": True},
format="json",
)
assert response.status_code == 201
duplicated_document = models.Document.objects.get(id=response.json()["id"])
assert duplicated_document.title == "Copy of document with accesses"
assert duplicated_document.content == document.content
assert duplicated_document.link_reach == document.link_reach
assert duplicated_document.link_role == document.link_role
assert duplicated_document.creator == user
assert duplicated_document.duplicated_from == document
assert duplicated_document.attachments == []
# Check that accesses were duplicated and the user who did the duplicate is forced as owner
duplicated_accesses = duplicated_document.accesses
assert duplicated_accesses.count() == 3
assert duplicated_accesses.get(user=user).role == "owner"
assert duplicated_accesses.get(user=user_access.user).role == user_access.role
assert duplicated_accesses.get(team=team_access.team).role == team_access.role
@@ -37,11 +37,9 @@ def test_api_documents_media_auth_unkown_document():
def test_api_documents_media_auth_anonymous_public():
"""Anonymous users should be able to retrieve attachments linked to a public document"""
document = factories.DocumentFactory(link_reach="public")
document_id = uuid4()
filename = f"{uuid4()!s}.jpg"
key = f"{document.pk!s}/attachments/{filename:s}"
key = f"{document_id!s}/attachments/{filename:s}"
default_storage.connection.meta.client.put_object(
Bucket=default_storage.bucket_name,
Key=key,
@@ -49,6 +47,8 @@ def test_api_documents_media_auth_anonymous_public():
ContentType="text/plain",
)
factories.DocumentFactory(id=document_id, link_reach="public", attachments=[key])
original_url = f"http://localhost/media/{key:s}"
response = APIClient().get(
"/api/v1.0/documents/media-auth/", HTTP_X_ORIGINAL_URL=original_url
@@ -81,8 +81,6 @@ def test_api_documents_media_auth_anonymous_public():
def test_api_documents_media_auth_extensions():
"""Files with extensions of any format should work."""
document = factories.DocumentFactory(link_reach="public")
extensions = [
"c",
"go",
@@ -91,10 +89,15 @@ def test_api_documents_media_auth_extensions():
"woff2",
"appimage",
]
document_id = uuid4()
keys = []
for ext in extensions:
filename = f"{uuid.uuid4()!s}.{ext:s}"
key = f"{document.pk!s}/attachments/{filename:s}"
filename = f"{uuid4()!s}.{ext:s}"
keys.append(f"{document_id!s}/attachments/{filename:s}")
factories.DocumentFactory(link_reach="public", attachments=keys)
for key in keys:
original_url = f"http://localhost/media/{key:s}"
response = APIClient().get(
"/api/v1.0/documents/media-auth/", HTTP_X_ORIGINAL_URL=original_url
@@ -109,10 +112,11 @@ def test_api_documents_media_auth_anonymous_authenticated_or_restricted(reach):
Anonymous users should not be allowed to retrieve attachments linked to a document
with link reach set to authenticated or restricted.
"""
document = factories.DocumentFactory(link_reach=reach)
document_id = uuid4()
filename = f"{uuid4()!s}.jpg"
media_url = f"http://localhost/media/{document.pk!s}/attachments/{filename:s}"
media_url = f"http://localhost/media/{document_id!s}/attachments/{filename:s}"
factories.DocumentFactory(id=document_id, link_reach=reach)
response = APIClient().get(
"/api/v1.0/documents/media-auth/", HTTP_X_ORIGINAL_URL=media_url
@@ -122,20 +126,16 @@ def test_api_documents_media_auth_anonymous_authenticated_or_restricted(reach):
assert "Authorization" not in response
@pytest.mark.parametrize("reach", ["public", "authenticated"])
def test_api_documents_media_auth_authenticated_public_or_authenticated(reach):
def test_api_documents_media_auth_anonymous_attachments():
"""
Authenticated users who are not related to a document should be able to retrieve
attachments related to a document with public or authenticated link reach.
Declaring a media key as original attachment on a document to which
a user has access should give them access to the attachment file
regardless of their access rights on the original document.
"""
document = factories.DocumentFactory(link_reach=reach)
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
document_id = uuid4()
filename = f"{uuid4()!s}.jpg"
key = f"{document.pk!s}/attachments/{filename:s}"
key = f"{document_id!s}/attachments/{filename:s}"
media_url = f"http://localhost/media/{key:s}"
default_storage.connection.meta.client.put_object(
Bucket=default_storage.bucket_name,
@@ -144,9 +144,73 @@ def test_api_documents_media_auth_authenticated_public_or_authenticated(reach):
ContentType="text/plain",
)
original_url = f"http://localhost/media/{key:s}"
factories.DocumentFactory(id=document_id, link_reach="restricted")
response = APIClient().get(
"/api/v1.0/documents/media-auth/", HTTP_X_ORIGINAL_URL=media_url
)
assert response.status_code == 403
# Let's now add a document to which the anonymous user has access and
# pointing to the attachment
parent = factories.DocumentFactory(link_reach="public")
factories.DocumentFactory(parent=parent, link_reach="restricted", attachments=[key])
response = APIClient().get(
"/api/v1.0/documents/media-auth/", HTTP_X_ORIGINAL_URL=media_url
)
assert response.status_code == 200
authorization = response["Authorization"]
assert "AWS4-HMAC-SHA256 Credential=" in authorization
assert (
"SignedHeaders=host;x-amz-content-sha256;x-amz-date, Signature="
in authorization
)
assert response["X-Amz-Date"] == timezone.now().strftime("%Y%m%dT%H%M%SZ")
s3_url = urlparse(settings.AWS_S3_ENDPOINT_URL)
file_url = f"{settings.AWS_S3_ENDPOINT_URL:s}/impress-media-storage/{key:s}"
response = requests.get(
file_url,
headers={
"authorization": authorization,
"x-amz-date": response["x-amz-date"],
"x-amz-content-sha256": response["x-amz-content-sha256"],
"Host": f"{s3_url.hostname:s}:{s3_url.port:d}",
},
timeout=1,
)
assert response.content.decode("utf-8") == "my prose"
@pytest.mark.parametrize("reach", ["public", "authenticated"])
def test_api_documents_media_auth_authenticated_public_or_authenticated(reach):
"""
Authenticated users who are not related to a document should be able to retrieve
attachments related to a document with public or authenticated link reach.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
document_id = uuid4()
filename = f"{uuid4()!s}.jpg"
key = f"{document_id!s}/attachments/{filename:s}"
media_url = f"http://localhost/media/{key:s}"
default_storage.connection.meta.client.put_object(
Bucket=default_storage.bucket_name,
Key=key,
Body=BytesIO(b"my prose"),
ContentType="text/plain",
)
factories.DocumentFactory(id=document_id, link_reach=reach, attachments=[key])
response = client.get(
"/api/v1.0/documents/media-auth/", HTTP_X_ORIGINAL_URL=original_url
"/api/v1.0/documents/media-auth/", HTTP_X_ORIGINAL_URL=media_url
)
assert response.status_code == 200
@@ -179,14 +243,18 @@ def test_api_documents_media_auth_authenticated_restricted():
Authenticated users who are not related to a document should not be allowed to
retrieve attachments linked to a document that is restricted.
"""
document = factories.DocumentFactory(link_reach="restricted")
user = factories.UserFactory(with_owned_document=True)
client = APIClient()
client.force_login(user)
document_id = uuid4()
filename = f"{uuid4()!s}.jpg"
media_url = f"http://localhost/media/{document.pk!s}/attachments/{filename:s}"
key = f"{document_id!s}/attachments/{filename:s}"
media_url = f"http://localhost/media/{key:s}"
factories.DocumentFactory(
id=document_id, link_reach="restricted", attachments=[key]
)
response = client.get(
"/api/v1.0/documents/media-auth/", HTTP_X_ORIGINAL_URL=media_url
@@ -206,16 +274,10 @@ def test_api_documents_media_auth_related(via, mock_user_teams):
client = APIClient()
client.force_login(user)
document = factories.DocumentFactory()
if via == USER:
factories.UserDocumentAccessFactory(document=document, user=user)
elif via == TEAM:
mock_user_teams.return_value = ["lasuite", "unknown"]
factories.TeamDocumentAccessFactory(document=document, team="lasuite")
document_id = uuid4()
filename = f"{uuid4()!s}.jpg"
key = f"{document.pk!s}/attachments/{filename:s}"
key = f"{document_id!s}/attachments/{filename:s}"
media_url = f"http://localhost/media/{key:s}"
default_storage.connection.meta.client.put_object(
Bucket=default_storage.bucket_name,
Key=key,
@@ -223,9 +285,17 @@ def test_api_documents_media_auth_related(via, mock_user_teams):
ContentType="text/plain",
)
original_url = f"http://localhost/media/{key:s}"
document = factories.DocumentFactory(
id=document_id, link_reach="restricted", attachments=[key]
)
if via == USER:
factories.UserDocumentAccessFactory(document=document, user=user)
elif via == TEAM:
mock_user_teams.return_value = ["lasuite", "unknown"]
factories.TeamDocumentAccessFactory(document=document, team="lasuite")
response = client.get(
"/api/v1.0/documents/media-auth/", HTTP_X_ORIGINAL_URL=original_url
"/api/v1.0/documents/media-auth/", HTTP_X_ORIGINAL_URL=media_url
)
assert response.status_code == 200
@@ -37,6 +37,7 @@ def test_api_documents_retrieve_anonymous_public_standalone():
"cors_proxy": True,
"descendants": True,
"destroy": False,
"duplicate": True,
# Anonymous user can't favorite a document even with read access
"favorite": False,
"invite_owner": False,
@@ -103,6 +104,7 @@ def test_api_documents_retrieve_anonymous_public_parent():
"descendants": True,
"cors_proxy": True,
"destroy": False,
"duplicate": True,
# Anonymous user can't favorite a document even with read access
"favorite": False,
"invite_owner": False,
@@ -198,6 +200,7 @@ def test_api_documents_retrieve_authenticated_unrelated_public_or_authenticated(
"descendants": True,
"cors_proxy": True,
"destroy": False,
"duplicate": True,
"favorite": True,
"invite_owner": False,
"link_configuration": False,
@@ -271,6 +274,7 @@ def test_api_documents_retrieve_authenticated_public_or_authenticated_parent(rea
"descendants": True,
"cors_proxy": True,
"destroy": False,
"duplicate": True,
"favorite": True,
"invite_owner": False,
"link_configuration": False,
@@ -450,6 +454,7 @@ def test_api_documents_retrieve_authenticated_related_parent():
"descendants": True,
"cors_proxy": True,
"destroy": access.role == "owner",
"duplicate": True,
"favorite": True,
"invite_owner": access.role == "owner",
"link_configuration": access.role in ["administrator", "owner"],
@@ -81,6 +81,7 @@ def test_api_documents_trashbin_format():
"descendants": True,
"cors_proxy": True,
"destroy": True,
"duplicate": True,
"favorite": True,
"invite_owner": True,
"link_configuration": True,
@@ -0,0 +1,153 @@
"""
Test extract-attachments on document update in docs core app.
"""
import base64
from uuid import uuid4
import pytest
import y_py
from rest_framework.test import APIClient
from core import factories
pytestmark = pytest.mark.django_db
def get_ydoc_with_mages(image_keys):
"""Return a ydoc from text for testing purposes."""
ydoc = y_py.YDoc() # pylint: disable=no-member
with ydoc.begin_transaction() as txn:
xml_fragment = ydoc.get_xml_element("document-store")
for key in image_keys:
xml_image = xml_fragment.push_xml_element(txn, "image")
xml_image.set_attribute(txn, "src", f"http://localhost/media/{key:s}")
update = y_py.encode_state_as_update(ydoc) # pylint: disable=no-member
return base64.b64encode(update).decode("utf-8")
def test_api_documents_update_new_attachment_keys_anonymous(django_assert_num_queries):
"""
When an anonymous user updates a document, the attachment keys extracted from the
updated content should be added to the list of "attachments" ot the document if these
attachments are already readable by anonymous users.
"""
image_keys = [f"{uuid4()!s}/attachments/{uuid4()!s}.png" for _ in range(4)]
document = factories.DocumentFactory(
content=get_ydoc_with_mages(image_keys[:1]),
attachments=[image_keys[0]],
link_reach="public",
link_role="editor",
)
factories.DocumentFactory(attachments=[image_keys[1]], link_reach="public")
factories.DocumentFactory(attachments=[image_keys[2]], link_reach="authenticated")
factories.DocumentFactory(attachments=[image_keys[3]], link_reach="restricted")
expected_keys = {image_keys[i] for i in [0, 1]}
with django_assert_num_queries(9):
response = APIClient().put(
f"/api/v1.0/documents/{document.id!s}/",
{"content": get_ydoc_with_mages(image_keys)},
format="json",
)
assert response.status_code == 200
document.refresh_from_db()
assert set(document.attachments) == expected_keys
# Check that the db query to check attachments readability for extracted
# keys is not done if the content changes but no new keys are found
with django_assert_num_queries(5):
response = APIClient().put(
f"/api/v1.0/documents/{document.id!s}/",
{"content": get_ydoc_with_mages(image_keys[:2])},
format="json",
)
assert response.status_code == 200
document.refresh_from_db()
assert len(document.attachments) == 2
assert set(document.attachments) == expected_keys
def test_api_documents_update_new_attachment_keys_authenticated(
django_assert_num_queries,
):
"""
When an authenticated user updates a document, the attachment keys extracted from the
updated content should be added to the list of "attachments" ot the document if these
attachments are already readable by the editing user.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
image_keys = [f"{uuid4()!s}/attachments/{uuid4()!s}.png" for _ in range(5)]
document = factories.DocumentFactory(
content=get_ydoc_with_mages(image_keys[:1]),
attachments=[image_keys[0]],
users=[(user, "editor")],
)
factories.DocumentFactory(attachments=[image_keys[1]], link_reach="public")
factories.DocumentFactory(attachments=[image_keys[2]], link_reach="authenticated")
factories.DocumentFactory(attachments=[image_keys[3]], link_reach="restricted")
factories.DocumentFactory(attachments=[image_keys[4]], users=[user])
expected_keys = {image_keys[i] for i in [0, 1, 2, 4]}
with django_assert_num_queries(10):
response = client.put(
f"/api/v1.0/documents/{document.id!s}/",
{"content": get_ydoc_with_mages(image_keys)},
format="json",
)
assert response.status_code == 200
document.refresh_from_db()
assert set(document.attachments) == expected_keys
# Check that the db query to check attachments readability for extracted
# keys is not done if the content changes but no new keys are found
with django_assert_num_queries(6):
response = client.put(
f"/api/v1.0/documents/{document.id!s}/",
{"content": get_ydoc_with_mages(image_keys[:2])},
format="json",
)
assert response.status_code == 200
document.refresh_from_db()
assert len(document.attachments) == 4
assert set(document.attachments) == expected_keys
def test_api_documents_update_new_attachment_keys_duplicate():
"""
Duplicate keys in the content should not result in duplicates in the document's attachments.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
image_key1 = f"{uuid4()!s}/attachments/{uuid4()!s}.png"
image_key2 = f"{uuid4()!s}/attachments/{uuid4()!s}.png"
document = factories.DocumentFactory(
content=get_ydoc_with_mages([image_key1]),
attachments=[image_key1],
users=[(user, "editor")],
)
factories.DocumentFactory(attachments=[image_key2], users=[user])
response = client.put(
f"/api/v1.0/documents/{document.id!s}/",
{"content": get_ydoc_with_mages([image_key1, image_key2, image_key2])},
format="json",
)
assert response.status_code == 200
document.refresh_from_db()
assert len(document.attachments) == 2
assert set(document.attachments) == {image_key1, image_key2}