diff --git a/swh/deposit/tests/api/test_collection_post_multipart.py b/swh/deposit/tests/api/test_collection_post_multipart.py --- a/swh/deposit/tests/api/test_collection_post_multipart.py +++ b/swh/deposit/tests/api/test_collection_post_multipart.py @@ -16,44 +16,29 @@ from swh.deposit.config import COL_IRI, DEPOSIT_STATUS_DEPOSITED from swh.deposit.models import Deposit, DepositRequest from swh.deposit.parsers import parse_xml -from swh.deposit.tests.common import check_archive +from swh.deposit.tests.common import check_archive, post_multipart def test_post_deposit_multipart_without_slug_header( - authenticated_client, deposit_collection, atom_dataset, mocker, deposit_user + authenticated_client, + deposit_collection, + atom_dataset, + mocker, + deposit_user, + sample_archive, ): # given url = reverse(COL_IRI, args=[deposit_collection.name]) - + data_atom_entry = atom_dataset["entry-data-deposit-binary"] id_ = str(uuid.uuid4()) mocker.patch("uuid.uuid4", return_value=id_) - archive_content = b"some content representing archive" - archive = InMemoryUploadedFile( - BytesIO(archive_content), - field_name="archive0", - name="archive0", - content_type="application/zip", - size=len(archive_content), - charset=None, - ) - - data_atom_entry = atom_dataset["entry-data-deposit-binary"] - atom_entry = InMemoryUploadedFile( - BytesIO(data_atom_entry.encode("utf-8")), - field_name="atom0", - name="atom0", - content_type='application/atom+xml; charset="utf-8"', - size=len(data_atom_entry), - charset="utf-8", - ) - # when - response = authenticated_client.post( + response = post_multipart( + authenticated_client, url, - format="multipart", - data={"archive": archive, "atom_entry": atom_entry,}, - # + headers + sample_archive, + data_atom_entry, HTTP_IN_PROGRESS="false", ) @@ -75,34 +60,15 @@ """ # given url = reverse(COL_IRI, args=[deposit_collection.name]) - - archive = InMemoryUploadedFile( - BytesIO(sample_archive["data"]), - field_name=sample_archive["name"], - name=sample_archive["name"], - content_type="application/zip", - size=sample_archive["length"], - charset=None, - ) - data_atom_entry = atom_dataset["entry-data-deposit-binary"] - atom_entry = InMemoryUploadedFile( - BytesIO(data_atom_entry.encode("utf-8")), - field_name="atom0", - name="atom0", - content_type='application/atom+xml; charset="utf-8"', - size=len(data_atom_entry), - charset="utf-8", - ) - external_id = "external-id" # when - response = authenticated_client.post( + response = post_multipart( + authenticated_client, url, - format="multipart", - data={"archive": archive, "atom_entry": atom_entry,}, - # + headers + sample_archive, + data_atom_entry, HTTP_IN_PROGRESS="false", HTTP_SLUG=external_id, ) @@ -143,36 +109,15 @@ """ # given url = reverse(COL_IRI, args=[deposit_collection.name]) - - # from django.core.files import uploadedfile data_atom_entry = atom_dataset["entry-data-deposit-binary"] - - archive = InMemoryUploadedFile( - BytesIO(sample_archive["data"]), - field_name=sample_archive["name"], - name=sample_archive["name"], - content_type="application/x-tar", - size=sample_archive["length"], - charset=None, - ) - - atom_entry = InMemoryUploadedFile( - BytesIO(data_atom_entry.encode("utf-8")), - field_name="atom0", - name="atom0", - content_type='application/atom+xml; charset="utf-8"', - size=len(data_atom_entry), - charset="utf-8", - ) - external_id = "external-id" # when - response = authenticated_client.post( + response = post_multipart( + authenticated_client, url, - format="multipart", - data={"archive": archive, "atom_entry": atom_entry,}, - # + headers + sample_archive, + data_atom_entry, HTTP_IN_PROGRESS="false", HTTP_SLUG=external_id, ) @@ -214,35 +159,15 @@ """ # given url = reverse(COL_IRI, args=[deposit_collection.name]) - data_atom_entry = atom_dataset["entry-data-deposit-binary"] - - archive = InMemoryUploadedFile( - BytesIO(sample_archive["data"]), - field_name=sample_archive["name"], - name=sample_archive["name"], - content_type="application/zip", - size=sample_archive["length"], - charset=None, - ) - - atom_entry = InMemoryUploadedFile( - BytesIO(data_atom_entry.encode("utf-8")), - field_name="atom0", - name="atom0", - content_type='application/atom+xml; charset="utf-8"', - size=len(data_atom_entry), - charset="utf-8", - ) - external_id = "external-id" # when - response = authenticated_client.post( + response = post_multipart( + authenticated_client, url, - format="multipart", - data={"archive": archive, "atom_entry": atom_entry,}, - # + headers + sample_archive, + data_atom_entry, HTTP_IN_PROGRESS="true", HTTP_SLUG=external_id, ) @@ -379,33 +304,14 @@ ): # given url = reverse(COL_IRI, args=[deposit_collection.name]) - - archive_content = sample_archive["data"] - archive = InMemoryUploadedFile( - BytesIO(archive_content), - field_name=sample_archive["name"], - name=sample_archive["name"], - content_type="application/zip", - size=len(archive_content), - charset=None, - ) - data_atom_entry_ko = atom_dataset["entry-data-ko"] - atom_entry = InMemoryUploadedFile( - BytesIO(data_atom_entry_ko.encode("utf-8")), - field_name="atom0", - name="atom0", - content_type='application/atom+xml; charset="utf-8"', - size=len(data_atom_entry_ko), - charset="utf-8", - ) # when - response = authenticated_client.post( + response = post_multipart( + authenticated_client, url, - format="multipart", - data={"archive": archive, "atom_entry": atom_entry,}, - # + headers + sample_archive, + data_atom_entry_ko, HTTP_IN_PROGRESS="false", HTTP_SLUG="external-id", ) @@ -420,34 +326,20 @@ # given url = reverse(COL_IRI, args=[deposit_collection.name]) - data = sample_archive["data"] * 8 - archive = InMemoryUploadedFile( - BytesIO(data), - field_name=sample_archive["name"], - name=sample_archive["name"], - content_type="application/zip", - size=len(data), - charset=None, - ) - + archive = { + **sample_archive, + "data": sample_archive["data"] * 8, + } data_atom_entry = atom_dataset["entry-data-deposit-binary"] - atom_entry = InMemoryUploadedFile( - BytesIO(data_atom_entry.encode("utf-8")), - field_name="atom0", - name="atom0", - content_type='application/atom+xml; charset="utf-8"', - size=len(data_atom_entry), - charset="utf-8", - ) external_id = "external-id" # when - response = authenticated_client.post( + response = post_multipart( + authenticated_client, url, - format="multipart", - data={"archive": archive, "atom_entry": atom_entry,}, - # + headers + archive, + data_atom_entry, HTTP_IN_PROGRESS="false", HTTP_SLUG=external_id, ) diff --git a/swh/deposit/tests/api/test_deposit_update.py b/swh/deposit/tests/api/test_deposit_update.py --- a/swh/deposit/tests/api/test_deposit_update.py +++ b/swh/deposit/tests/api/test_deposit_update.py @@ -5,9 +5,6 @@ """Tests updates on SE-IRI.""" -from io import BytesIO - -from django.core.files.uploadedfile import InMemoryUploadedFile from django.urls import reverse from rest_framework import status @@ -18,7 +15,7 @@ SE_IRI, ) from swh.deposit.models import Deposit, DepositRequest -from swh.deposit.tests.common import post_atom +from swh.deposit.tests.common import post_atom, post_multipart, put_multipart def test_add_both_archive_and_metadata_to_deposit( @@ -42,31 +39,12 @@ requests_archive0 = DepositRequest.objects.filter(deposit=deposit, type="archive") assert len(requests_archive0) == 1 - update_uri = reverse(EDIT_IRI, args=[deposit_collection.name, deposit.id]) - archive = InMemoryUploadedFile( - BytesIO(sample_archive["data"]), - field_name=sample_archive["name"], - name=sample_archive["name"], - content_type="application/x-tar", - size=sample_archive["length"], - charset=None, - ) - data_atom_entry = atom_dataset["entry-data1"] - atom_entry = InMemoryUploadedFile( - BytesIO(data_atom_entry.encode("utf-8")), - field_name="atom0", - name="atom0", - content_type='application/atom+xml; charset="utf-8"', - size=len(data_atom_entry), - charset="utf-8", - ) - - update_uri = reverse(SE_IRI, args=[deposit_collection.name, deposit.id]) - response = authenticated_client.post( - update_uri, - format="multipart", - data={"archive": archive, "atom_entry": atom_entry,}, + response = post_multipart( + authenticated_client, + reverse(SE_IRI, args=[deposit_collection.name, deposit.id]), + sample_archive, + data_atom_entry, ) assert response.status_code == status.HTTP_201_CREATED @@ -137,30 +115,12 @@ requests_archive0 = DepositRequest.objects.filter(deposit=deposit, type="archive") assert len(requests_archive0) == 1 - archive = InMemoryUploadedFile( - BytesIO(sample_archive["data"]), - field_name=sample_archive["name"], - name=sample_archive["name"], - content_type="application/x-tar", - size=sample_archive["length"], - charset=None, - ) - data_atom_entry = atom_dataset["entry-data1"] - atom_entry = InMemoryUploadedFile( - BytesIO(data_atom_entry.encode("utf-8")), - field_name="atom0", - name="atom0", - content_type='application/atom+xml; charset="utf-8"', - size=len(data_atom_entry), - charset="utf-8", - ) - - update_uri = reverse(EDIT_IRI, args=[deposit_collection.name, deposit.id]) - response = authenticated_client.put( - update_uri, - format="multipart", - data={"archive": archive, "atom_entry": atom_entry,}, + response = put_multipart( + authenticated_client, + reverse(EDIT_IRI, args=[deposit_collection.name, deposit.id]), + sample_archive, + data_atom_entry, ) assert response.status_code == status.HTTP_204_NO_CONTENT diff --git a/swh/deposit/tests/common.py b/swh/deposit/tests/common.py --- a/swh/deposit/tests/common.py +++ b/swh/deposit/tests/common.py @@ -4,11 +4,14 @@ # See top-level LICENSE file for more information import hashlib +from io import BytesIO import os import re import tarfile import tempfile +from django.core.files.uploadedfile import InMemoryUploadedFile + from swh.core import tarball @@ -175,3 +178,38 @@ return authenticated_client.put( url, content_type="application/atom+xml;type=entry", data=data, **kwargs ) + + +def _post_or_put_multipart(f, url, archive, atom_entry, **kwargs): + archive = InMemoryUploadedFile( + BytesIO(archive["data"]), + field_name=archive["name"], + name=archive["name"], + content_type="application/x-tar", + size=archive["length"], + charset=None, + ) + + atom_entry = InMemoryUploadedFile( + BytesIO(atom_entry.encode("utf-8")), + field_name="atom0", + name="atom0", + content_type='application/atom+xml; charset="utf-8"', + size=len(atom_entry), + charset="utf-8", + ) + + return f( + url, + format="multipart", + data={"archive": archive, "atom_entry": atom_entry,}, + **kwargs, + ) + + +def post_multipart(authenticated_client, *args, **kwargs): + return _post_or_put_multipart(authenticated_client.post, *args, **kwargs) + + +def put_multipart(authenticated_client, *args, **kwargs): + return _post_or_put_multipart(authenticated_client.put, *args, **kwargs)