Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/tests/api/test_deposit_update.py
| # Copyright (C) 2017-2020 The Software Heritage developers | # Copyright (C) 2017-2020 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| from io import BytesIO | from io import BytesIO | ||||
| import attr | import attr | ||||
| from django.core.files.uploadedfile import InMemoryUploadedFile | from django.core.files.uploadedfile import InMemoryUploadedFile | ||||
| from django.urls import reverse | from django.urls import reverse | ||||
| from rest_framework import status | from rest_framework import status | ||||
| from swh.deposit.api.common import ACCEPT_ARCHIVE_CONTENT_TYPES | from swh.deposit.api.common import ACCEPT_ARCHIVE_CONTENT_TYPES | ||||
| from swh.deposit.config import ( | from swh.deposit.config import ( | ||||
| COL_IRI, | |||||
| DEPOSIT_STATUS_DEPOSITED, | DEPOSIT_STATUS_DEPOSITED, | ||||
| DEPOSIT_STATUS_PARTIAL, | DEPOSIT_STATUS_PARTIAL, | ||||
| EDIT_IRI, | EDIT_IRI, | ||||
| EM_IRI, | EM_IRI, | ||||
| SE_IRI, | SE_IRI, | ||||
| APIConfig, | APIConfig, | ||||
| ) | ) | ||||
| from swh.deposit.models import Deposit, DepositCollection, DepositRequest | from swh.deposit.models import Deposit, DepositCollection, DepositRequest | ||||
| ▲ Show 20 Lines • Show All 77 Lines • ▼ Show 20 Lines | |||||
| def test_replace_metadata_to_deposit_is_possible( | def test_replace_metadata_to_deposit_is_possible( | ||||
| tmp_path, | tmp_path, | ||||
| authenticated_client, | authenticated_client, | ||||
| partial_deposit_with_metadata, | partial_deposit_with_metadata, | ||||
| deposit_collection, | deposit_collection, | ||||
| atom_dataset, | atom_dataset, | ||||
| deposit_user, | |||||
| ): | ): | ||||
| """Replace all metadata with another one should return a 204 response | """Replace all metadata with another one should return a 204 response | ||||
| """ | """ | ||||
| # given | # given | ||||
| deposit = partial_deposit_with_metadata | deposit = partial_deposit_with_metadata | ||||
| raw_metadata0 = atom_dataset["entry-data0"] | origin_url = deposit_user.provider_url + deposit.external_id | ||||
| raw_metadata0 = atom_dataset["entry-data0"] % origin_url | |||||
| requests_meta = DepositRequest.objects.filter(deposit=deposit, type="metadata") | requests_meta = DepositRequest.objects.filter(deposit=deposit, type="metadata") | ||||
| assert len(requests_meta) == 1 | assert len(requests_meta) == 1 | ||||
| request_meta0 = requests_meta[0] | request_meta0 = requests_meta[0] | ||||
| assert request_meta0.raw_metadata == raw_metadata0 | assert request_meta0.raw_metadata == raw_metadata0 | ||||
| requests_archive0 = DepositRequest.objects.filter(deposit=deposit, type="archive") | requests_archive0 = DepositRequest.objects.filter(deposit=deposit, type="archive") | ||||
| assert len(requests_archive0) == 1 | assert len(requests_archive0) == 1 | ||||
| ▲ Show 20 Lines • Show All 82 Lines • ▼ Show 20 Lines | ): | ||||
| assert set(requests_meta0) == set(requests_meta1) | assert set(requests_meta0) == set(requests_meta1) | ||||
| def test_add_metadata_to_deposit_is_possible( | def test_add_metadata_to_deposit_is_possible( | ||||
| authenticated_client, | authenticated_client, | ||||
| deposit_collection, | deposit_collection, | ||||
| partial_deposit_with_metadata, | partial_deposit_with_metadata, | ||||
| atom_dataset, | atom_dataset, | ||||
| deposit_user, | |||||
| ): | ): | ||||
| """Add metadata with another one should return a 204 response | """Add metadata with another one should return a 204 response | ||||
| """ | """ | ||||
| deposit = partial_deposit_with_metadata | deposit = partial_deposit_with_metadata | ||||
| origin_url = deposit_user.provider_url + deposit.external_id | |||||
| requests = DepositRequest.objects.filter(deposit=deposit, type="metadata") | requests = DepositRequest.objects.filter(deposit=deposit, type="metadata") | ||||
| assert len(requests) == 1 | assert len(requests) == 1 | ||||
| requests_archive0 = DepositRequest.objects.filter(deposit=deposit, type="archive") | requests_archive0 = DepositRequest.objects.filter(deposit=deposit, type="archive") | ||||
| assert len(requests_archive0) == 1 | assert len(requests_archive0) == 1 | ||||
| update_uri = reverse(SE_IRI, args=[deposit_collection.name, deposit.id]) | update_uri = reverse(SE_IRI, args=[deposit_collection.name, deposit.id]) | ||||
| atom_entry = atom_dataset["entry-data1"] | atom_entry = atom_dataset["entry-data1"] | ||||
| response = authenticated_client.post( | response = authenticated_client.post( | ||||
| update_uri, content_type="application/atom+xml;type=entry", data=atom_entry | update_uri, content_type="application/atom+xml;type=entry", data=atom_entry | ||||
| ) | ) | ||||
| assert response.status_code == status.HTTP_201_CREATED | assert response.status_code == status.HTTP_201_CREATED | ||||
| requests = DepositRequest.objects.filter(deposit=deposit, type="metadata").order_by( | requests = DepositRequest.objects.filter(deposit=deposit, type="metadata").order_by( | ||||
| "id" | "id" | ||||
| ) | ) | ||||
| assert len(requests) == 2 | assert len(requests) == 2 | ||||
| expected_raw_meta0 = atom_dataset["entry-data0"] | expected_raw_meta0 = atom_dataset["entry-data0"] % origin_url | ||||
| # a new one was added | # a new one was added | ||||
| assert requests[0].raw_metadata == expected_raw_meta0 | assert requests[0].raw_metadata == expected_raw_meta0 | ||||
| assert requests[1].raw_metadata == atom_entry | assert requests[1].raw_metadata == atom_entry | ||||
| # check we did not touch the other parts | # check we did not touch the other parts | ||||
| requests_archive1 = DepositRequest.objects.filter(deposit=deposit, type="archive") | requests_archive1 = DepositRequest.objects.filter(deposit=deposit, type="archive") | ||||
| assert len(requests_archive1) == 1 | assert len(requests_archive1) == 1 | ||||
| assert set(requests_archive0) == set(requests_archive1) | assert set(requests_archive0) == set(requests_archive1) | ||||
| def test_add_both_archive_and_metadata_to_deposit( | def test_add_both_archive_and_metadata_to_deposit( | ||||
| authenticated_client, | authenticated_client, | ||||
| deposit_collection, | deposit_collection, | ||||
| partial_deposit_with_metadata, | partial_deposit_with_metadata, | ||||
| atom_dataset, | atom_dataset, | ||||
| sample_archive, | sample_archive, | ||||
| deposit_user, | |||||
| ): | ): | ||||
| """Scenario: Add both a new archive and new metadata to a partial deposit is ok | """Scenario: Add both a new archive and new metadata to a partial deposit is ok | ||||
| Response: 201 | Response: 201 | ||||
| """ | """ | ||||
| deposit = partial_deposit_with_metadata | deposit = partial_deposit_with_metadata | ||||
| origin_url = deposit_user.provider_url + deposit.external_id | |||||
| requests = DepositRequest.objects.filter(deposit=deposit, type="metadata") | requests = DepositRequest.objects.filter(deposit=deposit, type="metadata") | ||||
| assert len(requests) == 1 | assert len(requests) == 1 | ||||
| requests_archive0 = DepositRequest.objects.filter(deposit=deposit, type="archive") | requests_archive0 = DepositRequest.objects.filter(deposit=deposit, type="archive") | ||||
| assert len(requests_archive0) == 1 | assert len(requests_archive0) == 1 | ||||
| update_uri = reverse(EDIT_IRI, args=[deposit_collection.name, deposit.id]) | update_uri = reverse(EDIT_IRI, args=[deposit_collection.name, deposit.id]) | ||||
| archive = InMemoryUploadedFile( | archive = InMemoryUploadedFile( | ||||
| Show All 23 Lines | ): | ||||
| ) | ) | ||||
| assert response.status_code == status.HTTP_201_CREATED | assert response.status_code == status.HTTP_201_CREATED | ||||
| requests = DepositRequest.objects.filter(deposit=deposit, type="metadata").order_by( | requests = DepositRequest.objects.filter(deposit=deposit, type="metadata").order_by( | ||||
| "id" | "id" | ||||
| ) | ) | ||||
| assert len(requests) == 1 + 1, "New deposit request archive got added" | assert len(requests) == 1 + 1, "New deposit request archive got added" | ||||
| expected_raw_meta0 = atom_dataset["entry-data0"] | expected_raw_meta0 = atom_dataset["entry-data0"] % origin_url | ||||
| # a new one was added | # a new one was added | ||||
| assert requests[0].raw_metadata == expected_raw_meta0 | assert requests[0].raw_metadata == expected_raw_meta0 | ||||
| assert requests[1].raw_metadata == data_atom_entry | assert requests[1].raw_metadata == data_atom_entry | ||||
| # check we did not touch the other parts | # check we did not touch the other parts | ||||
| requests_archive1 = DepositRequest.objects.filter(deposit=deposit, type="archive") | requests_archive1 = DepositRequest.objects.filter(deposit=deposit, type="archive") | ||||
| assert len(requests_archive1) == 1 + 1, "New deposit request metadata got added" | assert len(requests_archive1) == 1 + 1, "New deposit request metadata got added" | ||||
| ▲ Show 20 Lines • Show All 192 Lines • ▼ Show 20 Lines | |||||
| def test_put_update_metadata_and_archive_deposit_partial_nominal( | def test_put_update_metadata_and_archive_deposit_partial_nominal( | ||||
| tmp_path, | tmp_path, | ||||
| authenticated_client, | authenticated_client, | ||||
| partial_deposit_with_metadata, | partial_deposit_with_metadata, | ||||
| deposit_collection, | deposit_collection, | ||||
| atom_dataset, | atom_dataset, | ||||
| sample_archive, | sample_archive, | ||||
| deposit_user, | |||||
| ): | ): | ||||
| """Scenario: Replace metadata and archive(s) with new ones should be ok | """Scenario: Replace metadata and archive(s) with new ones should be ok | ||||
| Response: 204 | Response: 204 | ||||
| """ | """ | ||||
| # given | # given | ||||
| deposit = partial_deposit_with_metadata | deposit = partial_deposit_with_metadata | ||||
| raw_metadata0 = atom_dataset["entry-data0"] | origin_url = deposit_user.provider_url + deposit.external_id | ||||
| raw_metadata0 = atom_dataset["entry-data0"] % origin_url | |||||
| requests_meta = DepositRequest.objects.filter(deposit=deposit, type="metadata") | requests_meta = DepositRequest.objects.filter(deposit=deposit, type="metadata") | ||||
| assert len(requests_meta) == 1 | assert len(requests_meta) == 1 | ||||
| request_meta0 = requests_meta[0] | request_meta0 = requests_meta[0] | ||||
| assert request_meta0.raw_metadata == raw_metadata0 | assert request_meta0.raw_metadata == raw_metadata0 | ||||
| requests_archive0 = DepositRequest.objects.filter(deposit=deposit, type="archive") | requests_archive0 = DepositRequest.objects.filter(deposit=deposit, type="archive") | ||||
| assert len(requests_archive0) == 1 | assert len(requests_archive0) == 1 | ||||
| ▲ Show 20 Lines • Show All 259 Lines • ▼ Show 20 Lines | ): | ||||
| assert response.status_code == status.HTTP_400_BAD_REQUEST | assert response.status_code == status.HTTP_400_BAD_REQUEST | ||||
| assert b"Functional metadata checks failure" in response.content | assert b"Functional metadata checks failure" in response.content | ||||
| # detail on the errors | # detail on the errors | ||||
| assert b"- Mandatory fields are missing (atom:author)" in response.content | assert b"- Mandatory fields are missing (atom:author)" in response.content | ||||
| assert ( | assert ( | ||||
| b"- Mandatory alternate fields are missing (atom:name or atom:title)" | b"- Mandatory alternate fields are missing (atom:name or atom:title)" | ||||
| in response.content | in response.content | ||||
| ) | ) | ||||
| def test_put_atom_with_create_origin_and_external_identifier( | |||||
| authenticated_client, deposit_collection, atom_dataset, deposit_user | |||||
| ): | |||||
| """<atom:external_identifier> was deprecated before <swh:create_origin> | |||||
| was introduced, clients should get an error when trying to use both | |||||
| """ | |||||
| external_id = "foobar" | |||||
| origin_url = deposit_user.provider_url + external_id | |||||
| url = reverse(COL_IRI, args=[deposit_collection.name]) | |||||
| response = authenticated_client.post( | |||||
| url, | |||||
| content_type="application/atom+xml;type=entry", | |||||
| data=atom_dataset["entry-data0"] % origin_url, | |||||
| HTTP_IN_PROGRESS="true", | |||||
| ) | |||||
| assert response.status_code == status.HTTP_201_CREATED | |||||
| response_content = parse_xml(BytesIO(response.content)) | |||||
| for link in response_content["atom:link"]: | |||||
| if link["@rel"] == "edit": | |||||
| edit_iri = link["@href"] | |||||
| break | |||||
| else: | |||||
| assert False, response_content | |||||
| # when | |||||
| response = authenticated_client.put( | |||||
| edit_iri, | |||||
| content_type="application/atom+xml;type=entry", | |||||
| data=atom_dataset["error-with-external-identifier"] % external_id, | |||||
| # + headers | |||||
| HTTP_IN_PROGRESS="false", | |||||
| ) | |||||
| assert b"<external_identifier> is deprecated" in response.content | |||||
| assert response.status_code == status.HTTP_400_BAD_REQUEST | |||||
| def test_put_atom_with_create_origin_and_reference( | |||||
| authenticated_client, deposit_collection, atom_dataset, deposit_user | |||||
| ): | |||||
| """<swh:reference> and <swh:create_origin> are mutually exclusive | |||||
| """ | |||||
| external_id = "foobar" | |||||
| origin_url = deposit_user.provider_url + external_id | |||||
| url = reverse(COL_IRI, args=[deposit_collection.name]) | |||||
| response = authenticated_client.post( | |||||
| url, | |||||
| content_type="application/atom+xml;type=entry", | |||||
| data=atom_dataset["entry-data0"] % origin_url, | |||||
| HTTP_IN_PROGRESS="true", | |||||
| ) | |||||
| assert response.status_code == status.HTTP_201_CREATED | |||||
| response_content = parse_xml(BytesIO(response.content)) | |||||
| for link in response_content["atom:link"]: | |||||
| if link["@rel"] == "edit": | |||||
| edit_iri = link["@href"] | |||||
| break | |||||
| else: | |||||
| assert False, response_content | |||||
| # when | |||||
| response = authenticated_client.put( | |||||
| edit_iri, | |||||
| content_type="application/atom+xml;type=entry", | |||||
| data=atom_dataset["entry-data-with-origin-reference"].format(url=origin_url), | |||||
| # + headers | |||||
| HTTP_IN_PROGRESS="false", | |||||
| ) | |||||
| assert b"only one may be used on a given deposit" in response.content | |||||
| assert response.status_code == status.HTTP_400_BAD_REQUEST | |||||