Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/tests/api/test_collection_post_multipart.py
Show All 11 Lines | |||||
from django.urls import reverse_lazy as reverse | from django.urls import reverse_lazy as 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, post_multipart | from swh.deposit.tests.common import check_archive, post_multipart | ||||
from swh.deposit.utils import NAMESPACES | |||||
def test_post_deposit_multipart( | def test_post_deposit_multipart( | ||||
authenticated_client, | authenticated_client, | ||||
deposit_collection, | deposit_collection, | ||||
atom_dataset, | atom_dataset, | ||||
mocker, | mocker, | ||||
deposit_user, | deposit_user, | ||||
Show All 10 Lines | response = post_multipart( | ||||
authenticated_client, | authenticated_client, | ||||
url, | url, | ||||
sample_archive, | sample_archive, | ||||
data_atom_entry, | data_atom_entry, | ||||
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(response.content) | ||||
deposit_id = response_content["swh:deposit_id"] | deposit_id = int(response_content.findtext("swh:deposit_id", namespaces=NAMESPACES)) | ||||
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 == origin_url | assert deposit.origin_url == origin_url | ||||
assert deposit.status == DEPOSIT_STATUS_DEPOSITED | assert deposit.status == DEPOSIT_STATUS_DEPOSITED | ||||
def test_post_deposit_multipart_without_origin_url( | def test_post_deposit_multipart_without_origin_url( | ||||
Show All 15 Lines | response = post_multipart( | ||||
authenticated_client, | authenticated_client, | ||||
url, | url, | ||||
sample_archive, | sample_archive, | ||||
data_atom_entry, | data_atom_entry, | ||||
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(response.content) | ||||
deposit_id = response_content["swh:deposit_id"] | deposit_id = int(response_content.findtext("swh:deposit_id", namespaces=NAMESPACES)) | ||||
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( | ||||
Show All 15 Lines | response = post_multipart( | ||||
data_atom_entry, | data_atom_entry, | ||||
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(response.content) | ||||
deposit_id = response_content["swh:deposit_id"] | deposit_id = int(response_content.findtext("swh:deposit_id", namespaces=NAMESPACES)) | ||||
deposit = Deposit.objects.get(pk=deposit_id) | deposit = Deposit.objects.get(pk=deposit_id) | ||||
assert deposit.status == DEPOSIT_STATUS_DEPOSITED | assert deposit.status == DEPOSIT_STATUS_DEPOSITED | ||||
assert deposit.external_id == external_id | assert deposit.external_id == external_id | ||||
assert deposit.collection == deposit_collection | assert deposit.collection == deposit_collection | ||||
assert deposit.swhid is None | assert deposit.swhid is None | ||||
deposit_requests = DepositRequest.objects.filter(deposit=deposit) | deposit_requests = DepositRequest.objects.filter(deposit=deposit) | ||||
assert len(deposit_requests) == 2 | assert len(deposit_requests) == 2 | ||||
for deposit_request in deposit_requests: | for deposit_request in deposit_requests: | ||||
assert deposit_request.deposit == deposit | assert deposit_request.deposit == deposit | ||||
if deposit_request.type == "archive": | if deposit_request.type == "archive": | ||||
check_archive(sample_archive["name"], deposit_request.archive.name) | check_archive(sample_archive["name"], deposit_request.archive.name) | ||||
assert deposit_request.metadata is None | |||||
assert deposit_request.raw_metadata is None | assert deposit_request.raw_metadata is None | ||||
else: | else: | ||||
assert ( | assert ( | ||||
deposit_request.metadata["atom:id"] | parse_xml(deposit_request.raw_metadata).findtext( | ||||
"atom:id", namespaces=NAMESPACES | |||||
) | |||||
== "urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a" | == "urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a" | ||||
) | ) | ||||
assert deposit_request.raw_metadata == data_atom_entry | assert deposit_request.raw_metadata == data_atom_entry | ||||
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 | ||||
): | ): | ||||
Show All 13 Lines | response = post_multipart( | ||||
data_atom_entry, | data_atom_entry, | ||||
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(response.content) | ||||
deposit_id = response_content["swh:deposit_id"] | deposit_id = int(response_content.findtext("swh:deposit_id", namespaces=NAMESPACES)) | ||||
deposit = Deposit.objects.get(pk=deposit_id) | deposit = Deposit.objects.get(pk=deposit_id) | ||||
assert deposit.status == DEPOSIT_STATUS_DEPOSITED | assert deposit.status == DEPOSIT_STATUS_DEPOSITED | ||||
assert deposit.external_id == external_id | assert deposit.external_id == external_id | ||||
assert deposit.collection == deposit_collection | assert deposit.collection == deposit_collection | ||||
assert deposit.swhid is None | assert deposit.swhid is None | ||||
deposit_requests = DepositRequest.objects.filter(deposit=deposit) | deposit_requests = DepositRequest.objects.filter(deposit=deposit) | ||||
assert len(deposit_requests) == 2 | assert len(deposit_requests) == 2 | ||||
for deposit_request in deposit_requests: | for deposit_request in deposit_requests: | ||||
assert deposit_request.deposit == deposit | assert deposit_request.deposit == deposit | ||||
if deposit_request.type == "archive": | if deposit_request.type == "archive": | ||||
check_archive(sample_archive["name"], deposit_request.archive.name) | check_archive(sample_archive["name"], deposit_request.archive.name) | ||||
assert deposit_request.metadata is None | |||||
assert deposit_request.raw_metadata is None | assert deposit_request.raw_metadata is None | ||||
else: | else: | ||||
assert ( | assert ( | ||||
deposit_request.metadata["atom:id"] | parse_xml(deposit_request.raw_metadata).findtext( | ||||
"atom:id", namespaces=NAMESPACES | |||||
) | |||||
== "urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a" | == "urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a" | ||||
) | ) | ||||
assert deposit_request.raw_metadata == data_atom_entry | assert deposit_request.raw_metadata == data_atom_entry | ||||
def test_post_deposit_multipart_put_to_replace_metadata( | 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 | ||||
): | ): | ||||
Show All 14 Lines | response = post_multipart( | ||||
data_atom_entry, | data_atom_entry, | ||||
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(response.content) | ||||
deposit_id = response_content["swh:deposit_id"] | deposit_id = int(response_content.findtext("swh:deposit_id", namespaces=NAMESPACES)) | ||||
deposit = Deposit.objects.get(pk=deposit_id) | deposit = Deposit.objects.get(pk=deposit_id) | ||||
assert deposit.status == "partial" | assert deposit.status == "partial" | ||||
assert deposit.external_id == external_id | assert deposit.external_id == external_id | ||||
assert deposit.collection == deposit_collection | assert deposit.collection == deposit_collection | ||||
assert deposit.swhid is None | assert deposit.swhid is None | ||||
deposit_requests = DepositRequest.objects.filter(deposit=deposit) | deposit_requests = DepositRequest.objects.filter(deposit=deposit) | ||||
assert len(deposit_requests) == 2 | assert len(deposit_requests) == 2 | ||||
for deposit_request in deposit_requests: | for deposit_request in deposit_requests: | ||||
assert deposit_request.deposit == deposit | assert deposit_request.deposit == deposit | ||||
if deposit_request.type == "archive": | if deposit_request.type == "archive": | ||||
check_archive(sample_archive["name"], deposit_request.archive.name) | check_archive(sample_archive["name"], deposit_request.archive.name) | ||||
else: | else: | ||||
assert ( | assert ( | ||||
deposit_request.metadata["atom:id"] | parse_xml(deposit_request.raw_metadata).findtext( | ||||
"atom:id", namespaces=NAMESPACES | |||||
) | |||||
== "urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a" | == "urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a" | ||||
) | ) | ||||
assert deposit_request.raw_metadata == data_atom_entry | assert deposit_request.raw_metadata == data_atom_entry | ||||
replace_metadata_uri = response["location"] | replace_metadata_uri = response["location"] | ||||
response = authenticated_client.put( | response = authenticated_client.put( | ||||
replace_metadata_uri, | replace_metadata_uri, | ||||
content_type="application/atom+xml;type=entry", | content_type="application/atom+xml;type=entry", | ||||
Show All 13 Lines | ): | ||||
deposit_requests = DepositRequest.objects.filter(deposit=deposit) | deposit_requests = DepositRequest.objects.filter(deposit=deposit) | ||||
assert len(deposit_requests) == 2 | assert len(deposit_requests) == 2 | ||||
for deposit_request in deposit_requests: | for deposit_request in deposit_requests: | ||||
assert deposit_request.deposit == deposit | assert deposit_request.deposit == deposit | ||||
if deposit_request.type == "archive": | if deposit_request.type == "archive": | ||||
check_archive(sample_archive["name"], deposit_request.archive.name) | check_archive(sample_archive["name"], deposit_request.archive.name) | ||||
else: | else: | ||||
assert ( | assert ( | ||||
deposit_request.metadata["atom:id"] | parse_xml(deposit_request.raw_metadata).findtext( | ||||
"atom:id", namespaces=NAMESPACES | |||||
) | |||||
== "urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a" | == "urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a" | ||||
) | ) | ||||
assert ( | assert ( | ||||
deposit_request.raw_metadata | deposit_request.raw_metadata | ||||
== atom_dataset["entry-data-deposit-binary"] | == atom_dataset["entry-data-deposit-binary"] | ||||
) | ) | ||||
▲ Show 20 Lines • Show All 118 Lines • Show Last 20 Lines |