Differential D4786 Diff 16946 swh/deposit/tests/api/test_collection_post_binary.py

Changeset View

Standalone View

swh/deposit/tests/api/test_collection_post_binary.py

Show All 9 Lines

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, create_arborescence_archive		from swh.deposit.tests.common import (
		check_archive,
		create_arborescence_archive,
		post_archive,
		)


def test_post_deposit_binary_no_slug(		def test_post_deposit_binary_no_slug(
authenticated_client, deposit_collection, sample_archive, deposit_user, mocker		authenticated_client, deposit_collection, sample_archive, deposit_user, mocker
):		):
"""Posting a binary deposit without slug header should generate one		"""Posting a binary deposit without slug header should generate one

"""		"""
id_ = str(uuid.uuid4())		id_ = str(uuid.uuid4())
mocker.patch("uuid.uuid4", return_value=id_)		mocker.patch("uuid.uuid4", return_value=id_)

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

# when		# when
response = authenticated_client.post(		response = post_archive(
url,		authenticated_client, url, sample_archive, in_progress="false",
content_type="application/zip", # as zip
data=sample_archive["data"],
# + headers
CONTENT_LENGTH=sample_archive["length"],
HTTP_CONTENT_MD5=sample_archive["md5sum"],
HTTP_PACKAGING="http://purl.org/net/sword/package/SimpleZip",
HTTP_IN_PROGRESS="false",
HTTP_CONTENT_DISPOSITION="attachment; filename=filename0",
)		)

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
Show All 10 Lines	):
# given		# given
url = reverse(COL_IRI, args=[deposit_collection.name])		url = reverse(COL_IRI, args=[deposit_collection.name])

external_id = "some-external-id-1"		external_id = "some-external-id-1"

# when		# when
response = authenticated_client.post(		response = authenticated_client.post(
url,		url,
content_type="application/octet-stream",		sample_archive,
data=sample_archive["data"],
# + headers
CONTENT_LENGTH=sample_archive["length"],
HTTP_SLUG=external_id,		HTTP_SLUG=external_id,
HTTP_CONTENT_MD5=sample_archive["md5sum"],		content_type="application/octet-stream",
HTTP_PACKAGING="http://purl.org/net/sword/package/SimpleZip",
HTTP_IN_PROGRESS="false",		HTTP_IN_PROGRESS="false",
HTTP_CONTENT_DISPOSITION="attachment; filename=filename0",
)		)

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

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


def test_post_deposit_binary_upload_ok(		def test_post_deposit_binary_upload_ok(
authenticated_client, deposit_collection, sample_archive		authenticated_client, deposit_collection, sample_archive
):		):
"""Binary upload with correct headers should return 201 with receipt		"""Binary upload with correct headers should return 201 with receipt

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

external_id = "some-external-id-1"		external_id = "some-external-id-1"

# when		# when
response = authenticated_client.post(		response = post_archive(
		authenticated_client,
url,		url,
content_type="application/zip", # as zip		sample_archive,
data=sample_archive["data"],
# + headers
CONTENT_LENGTH=sample_archive["length"],
# other headers needs HTTP_ prefix to be taken into account
HTTP_SLUG=external_id,		HTTP_SLUG=external_id,
HTTP_CONTENT_MD5=sample_archive["md5sum"],
HTTP_PACKAGING="http://purl.org/net/sword/package/SimpleZip",
HTTP_IN_PROGRESS="false",		HTTP_IN_PROGRESS="false",
HTTP_CONTENT_DISPOSITION="attachment; filename=%s" % (sample_archive["name"],),
)		)

# then		# then
response_content = parse_xml(BytesIO(response.content))		response_content = parse_xml(BytesIO(response.content))
assert response.status_code == status.HTTP_201_CREATED		assert response.status_code == status.HTTP_201_CREATED
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)
Show All 34 Lines	):

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

external_id = "some-external-id"		external_id = "some-external-id"

# when		# when
response = authenticated_client.post(		response = post_archive(
		authenticated_client,
url,		url,
content_type="application/zip",		sample_archive,
data=sample_archive["data"],
# + headers
CONTENT_LENGTH=sample_archive["length"],
HTTP_SLUG=external_id,		HTTP_SLUG=external_id,
HTTP_CONTENT_MD5=sample_archive["md5sum"],
HTTP_PACKAGING="something-unsupported",		HTTP_PACKAGING="something-unsupported",
HTTP_CONTENT_DISPOSITION="attachment; filename=filename0",
)		)

# then		# then
assert response.status_code == status.HTTP_400_BAD_REQUEST		assert response.status_code == status.HTTP_400_BAD_REQUEST
assert (		assert (
b"The packaging provided something-unsupported is not supported"		b"The packaging provided something-unsupported is not supported"
in response.content		in response.content
)		)
Show All 9 Lines	):

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

external_id = "some-external-id"		external_id = "some-external-id"

# when		# when
response = authenticated_client.post(		response = post_archive(
		authenticated_client,
url,		url,
content_type="application/zip",		sample_archive,
data=sample_archive["data"],
# + headers
CONTENT_LENGTH=sample_archive["length"],
HTTP_SLUG=external_id,		HTTP_SLUG=external_id,
HTTP_CONTENT_MD5=sample_archive["md5sum"],
HTTP_PACKAGING="http://purl.org/net/sword/package/SimpleZip",
HTTP_IN_PROGRESS="false",		HTTP_IN_PROGRESS="false",
		HTTP_CONTENT_DISPOSITION=None,
)		)

# then		# then
assert response.status_code == status.HTTP_400_BAD_REQUEST		assert response.status_code == status.HTTP_400_BAD_REQUEST
assert b"CONTENT_DISPOSITION header is mandatory" in response.content		assert b"CONTENT_DISPOSITION header is mandatory" 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)


def test_post_deposit_mediation_not_supported(		def test_post_deposit_mediation_not_supported(
authenticated_client, deposit_collection, sample_archive		authenticated_client, deposit_collection, sample_archive
):		):
"""Binary upload with mediation should return a 412 response		"""Binary upload with mediation should return a 412 response

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

external_id = "some-external-id-1"		external_id = "some-external-id-1"

# when		# when
response = authenticated_client.post(		response = post_archive(
		authenticated_client,
url,		url,
content_type="application/zip",		sample_archive,
data=sample_archive["data"],
# + headers
CONTENT_LENGTH=sample_archive["length"],
HTTP_SLUG=external_id,		HTTP_SLUG=external_id,
HTTP_CONTENT_MD5=sample_archive["md5sum"],
HTTP_PACKAGING="http://purl.org/net/sword/package/SimpleZip",
HTTP_IN_PROGRESS="false",		HTTP_IN_PROGRESS="false",
HTTP_ON_BEHALF_OF="someone",		HTTP_ON_BEHALF_OF="someone",
HTTP_CONTENT_DISPOSITION="attachment; filename=filename0",
)		)

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

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

Show All 9 Lines	):

archive = create_arborescence_archive(		archive = create_arborescence_archive(
tmp_path, "archive2", "file2", b"some content in file", up_to_size=500		tmp_path, "archive2", "file2", b"some content in file", up_to_size=500
)		)

external_id = "some-external-id"		external_id = "some-external-id"

# when		# when
response = authenticated_client.post(		response = post_archive(
		authenticated_client,
url,		url,
content_type="application/zip",		archive,
data=archive["data"],
# + headers
CONTENT_LENGTH=archive["length"],
HTTP_SLUG=external_id,		HTTP_SLUG=external_id,
HTTP_CONTENT_MD5=archive["md5sum"],
HTTP_PACKAGING="http://purl.org/net/sword/package/SimpleZip",
HTTP_IN_PROGRESS="false",		HTTP_IN_PROGRESS="false",
HTTP_CONTENT_DISPOSITION="attachment; filename=filename0",
)		)

# 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)
Show All 10 Lines	):

archive = create_arborescence_archive(		archive = create_arborescence_archive(
tmp_path, "archive2", "file2", b"some content in file", up_to_size=500		tmp_path, "archive2", "file2", b"some content in file", up_to_size=500
)		)

external_id = "some-external-id"		external_id = "some-external-id"

# when		# when
response = authenticated_client.post(		response = post_archive(
		authenticated_client,
url,		url,
content_type="application/zip",		archive,
data=archive["data"],
# + headers
CONTENT_LENGTH=None,		CONTENT_LENGTH=None,
HTTP_SLUG=external_id,		HTTP_SLUG=external_id,
HTTP_CONTENT_MD5=archive["md5sum"],
HTTP_PACKAGING="http://purl.org/net/sword/package/SimpleZip",
HTTP_IN_PROGRESS="false",		HTTP_IN_PROGRESS="false",
HTTP_CONTENT_DISPOSITION="attachment; filename=filename0",
)		)

# then		# then
assert response.status_code == status.HTTP_400_BAD_REQUEST		assert response.status_code == status.HTTP_400_BAD_REQUEST
assert b"the CONTENT_LENGTH header must be sent." in response.content		assert b"the CONTENT_LENGTH header must be sent." 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)


def test_post_deposit_2_post_2_different_deposits(		def test_post_deposit_2_post_2_different_deposits(
authenticated_client, deposit_collection, sample_archive		authenticated_client, deposit_collection, sample_archive
):		):
"""2 posting deposits should return 2 different 201 with receipt		"""2 posting deposits should return 2 different 201 with receipt

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

# when		# when
response = authenticated_client.post(		response = post_archive(
		authenticated_client,
url,		url,
content_type="application/zip", # as zip		sample_archive,
data=sample_archive["data"],
# + headers
CONTENT_LENGTH=sample_archive["length"],
HTTP_SLUG="some-external-id-1",		HTTP_SLUG="some-external-id-1",
HTTP_CONTENT_MD5=sample_archive["md5sum"],
HTTP_PACKAGING="http://purl.org/net/sword/package/SimpleZip",
HTTP_IN_PROGRESS="false",		HTTP_IN_PROGRESS="false",
HTTP_CONTENT_DISPOSITION="attachment; filename=filename0",
)		)

# 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))
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)

deposits = Deposit.objects.all()		deposits = Deposit.objects.all()
assert len(deposits) == 1		assert len(deposits) == 1
assert deposits[0] == deposit		assert deposits[0] == deposit

# second post		# second post
response = authenticated_client.post(		response = post_archive(
		authenticated_client,
url,		url,
content_type="application/x-tar", # as zip		sample_archive,
data=sample_archive["data"],		content_type="application/x-tar",
# + headers
CONTENT_LENGTH=sample_archive["length"],
HTTP_SLUG="another-external-id",		HTTP_SLUG="another-external-id",
HTTP_CONTENT_MD5=sample_archive["md5sum"],
HTTP_PACKAGING="http://purl.org/net/sword/package/SimpleZip",
HTTP_IN_PROGRESS="false",		HTTP_IN_PROGRESS="false",
HTTP_CONTENT_DISPOSITION="attachment; filename=filename1",
)		)

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_id2 = response_content["swh:deposit_id"]		deposit_id2 = response_content["swh:deposit_id"]

deposit2 = Deposit.objects.get(pk=deposit_id2)		deposit2 = Deposit.objects.get(pk=deposit_id2)

assert deposit != deposit2		assert deposit != deposit2

deposits = Deposit.objects.all().order_by("id")		deposits = Deposit.objects.all().order_by("id")
assert len(deposits) == 2		assert len(deposits) == 2
assert list(deposits), [deposit == deposit2]		assert list(deposits), [deposit == deposit2]