Differential D4788 Diff 17006 swh/deposit/tests/api/test_collection_post_multipart.py

Changeset View

Standalone View

swh/deposit/tests/api/test_collection_post_multipart.py

Show All 10 Lines
from django.core.files.uploadedfile import InMemoryUploadedFile		from django.core.files.uploadedfile import InMemoryUploadedFile
from django.urls import reverse		from django.urls import reverse
import pytest		import pytest
from rest_framework import status		from rest_framework import status

from swh.deposit.config import COL_IRI, DEPOSIT_STATUS_DEPOSITED		from swh.deposit.config import COL_IRI, DEPOSIT_STATUS_DEPOSITED
from swh.deposit.models import Deposit, DepositRequest		from swh.deposit.models import Deposit, DepositRequest
from swh.deposit.parsers import parse_xml		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(		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		# given
url = reverse(COL_IRI, args=[deposit_collection.name])		url = reverse(COL_IRI, args=[deposit_collection.name])
		data_atom_entry = atom_dataset["entry-data-deposit-binary"]
id_ = str(uuid.uuid4())		id_ = str(uuid.uuid4())
mocker.patch("uuid.uuid4", return_value=id_)		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		# when
response = authenticated_client.post(		response = post_multipart(
		authenticated_client,
url,		url,
format="multipart",		sample_archive,
data={"archive": archive, "atom_entry": atom_entry,},		data_atom_entry,
# + headers
HTTP_IN_PROGRESS="false",		HTTP_IN_PROGRESS="false",
)		)

assert response.status_code == status.HTTP_201_CREATED		assert response.status_code == status.HTTP_201_CREATED
response_content = parse_xml(BytesIO(response.content))		response_content = parse_xml(BytesIO(response.content))
deposit_id = response_content["swh:deposit_id"]		deposit_id = response_content["swh:deposit_id"]

deposit = Deposit.objects.get(pk=deposit_id)		deposit = Deposit.objects.get(pk=deposit_id)
assert deposit.collection == deposit_collection		assert deposit.collection == deposit_collection
assert deposit.origin_url == deposit_user.provider_url + id_		assert deposit.origin_url == deposit_user.provider_url + id_
assert deposit.status == DEPOSIT_STATUS_DEPOSITED		assert deposit.status == DEPOSIT_STATUS_DEPOSITED


def test_post_deposit_multipart_zip(		def test_post_deposit_multipart_zip(
authenticated_client, deposit_collection, atom_dataset, sample_archive		authenticated_client, deposit_collection, atom_dataset, sample_archive
):		):
"""one multipart deposit (zip+xml) should be accepted		"""one multipart deposit (zip+xml) should be accepted

"""		"""
# given		# given
url = reverse(COL_IRI, args=[deposit_collection.name])		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"]		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"		external_id = "external-id"

# when		# when
response = authenticated_client.post(		response = post_multipart(
		authenticated_client,
url,		url,
format="multipart",		sample_archive,
data={"archive": archive, "atom_entry": atom_entry,},		data_atom_entry,
# + headers
HTTP_IN_PROGRESS="false",		HTTP_IN_PROGRESS="false",
HTTP_SLUG=external_id,		HTTP_SLUG=external_id,
)		)

# then		# then
assert response.status_code == status.HTTP_201_CREATED		assert response.status_code == status.HTTP_201_CREATED

response_content = parse_xml(BytesIO(response.content))		response_content = parse_xml(BytesIO(response.content))
Show All 24 Lines
def test_post_deposit_multipart_tar(		def test_post_deposit_multipart_tar(
authenticated_client, deposit_collection, atom_dataset, sample_archive		authenticated_client, deposit_collection, atom_dataset, sample_archive
):		):
"""one multipart deposit (tar+xml) should be accepted		"""one multipart deposit (tar+xml) should be accepted

"""		"""
# given		# given
url = reverse(COL_IRI, args=[deposit_collection.name])		url = reverse(COL_IRI, args=[deposit_collection.name])

# from django.core.files import uploadedfile
data_atom_entry = atom_dataset["entry-data-deposit-binary"]		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"		external_id = "external-id"

# when		# when
response = authenticated_client.post(		response = post_multipart(
		authenticated_client,
url,		url,
format="multipart",		sample_archive,
data={"archive": archive, "atom_entry": atom_entry,},		data_atom_entry,
# + headers
HTTP_IN_PROGRESS="false",		HTTP_IN_PROGRESS="false",
HTTP_SLUG=external_id,		HTTP_SLUG=external_id,
)		)

# then		# then
assert response.status_code == status.HTTP_201_CREATED		assert response.status_code == status.HTTP_201_CREATED

response_content = parse_xml(BytesIO(response.content))		response_content = parse_xml(BytesIO(response.content))
Show All 25 Lines	def test_post_deposit_multipart_put_to_replace_metadata(
authenticated_client, deposit_collection, atom_dataset, sample_archive		authenticated_client, deposit_collection, atom_dataset, sample_archive
):		):
"""One multipart deposit followed by a metadata update should be		"""One multipart deposit followed by a metadata update should be
accepted		accepted

"""		"""
# given		# given
url = reverse(COL_IRI, args=[deposit_collection.name])		url = reverse(COL_IRI, args=[deposit_collection.name])

data_atom_entry = atom_dataset["entry-data-deposit-binary"]		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"		external_id = "external-id"

# when		# when
response = authenticated_client.post(		response = post_multipart(
		authenticated_client,
url,		url,
format="multipart",		sample_archive,
data={"archive": archive, "atom_entry": atom_entry,},		data_atom_entry,
# + headers
HTTP_IN_PROGRESS="true",		HTTP_IN_PROGRESS="true",
HTTP_SLUG=external_id,		HTTP_SLUG=external_id,
)		)

# then		# then
assert response.status_code == status.HTTP_201_CREATED		assert response.status_code == status.HTTP_201_CREATED

response_content = parse_xml(BytesIO(response.content))		response_content = parse_xml(BytesIO(response.content))
▲ Show 20 Lines • Show All 120 Lines • ▼ Show 20 Lines	):
) is True		) is True


def test_post_deposit_multipart_400_when_badly_formatted_xml(		def test_post_deposit_multipart_400_when_badly_formatted_xml(
authenticated_client, deposit_collection, sample_archive, atom_dataset		authenticated_client, deposit_collection, sample_archive, atom_dataset
):		):
# given		# given
url = reverse(COL_IRI, args=[deposit_collection.name])		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"]		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		# when
response = authenticated_client.post(		response = post_multipart(
		authenticated_client,
url,		url,
format="multipart",		sample_archive,
data={"archive": archive, "atom_entry": atom_entry,},		data_atom_entry_ko,
# + headers
HTTP_IN_PROGRESS="false",		HTTP_IN_PROGRESS="false",
HTTP_SLUG="external-id",		HTTP_SLUG="external-id",
)		)

assert b"Malformed xml metadata" in response.content		assert b"Malformed xml metadata" in response.content
assert response.status_code == status.HTTP_400_BAD_REQUEST		assert response.status_code == status.HTTP_400_BAD_REQUEST


def test_post_deposit_multipart_if_upload_size_limit_exceeded(		def test_post_deposit_multipart_if_upload_size_limit_exceeded(
authenticated_client, deposit_collection, atom_dataset, sample_archive		authenticated_client, deposit_collection, atom_dataset, sample_archive
):		):
# given		# given
url = reverse(COL_IRI, args=[deposit_collection.name])		url = reverse(COL_IRI, args=[deposit_collection.name])

data = sample_archive["data"] * 8		archive = {
archive = InMemoryUploadedFile(		**sample_archive,
BytesIO(data),		"data": sample_archive["data"] * 8,
field_name=sample_archive["name"],		}
name=sample_archive["name"],
content_type="application/zip",
size=len(data),
charset=None,
)

data_atom_entry = atom_dataset["entry-data-deposit-binary"]		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"		external_id = "external-id"

# when		# when
response = authenticated_client.post(		response = post_multipart(
		authenticated_client,
url,		url,
format="multipart",		archive,
data={"archive": archive, "atom_entry": atom_entry,},		data_atom_entry,
# + headers
HTTP_IN_PROGRESS="false",		HTTP_IN_PROGRESS="false",
HTTP_SLUG=external_id,		HTTP_SLUG=external_id,
)		)

# then		# then
assert response.status_code == status.HTTP_413_REQUEST_ENTITY_TOO_LARGE		assert response.status_code == status.HTTP_413_REQUEST_ENTITY_TOO_LARGE
assert b"Upload size limit exceeded" in response.content		assert b"Upload size limit exceeded" in response.content

with pytest.raises(Deposit.DoesNotExist):		with pytest.raises(Deposit.DoesNotExist):
Deposit.objects.get(external_id=external_id)		Deposit.objects.get(external_id=external_id)