Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/tests/api/test_collection.py
- This file was copied to swh/deposit/tests/api/test_collection_reuse_slug.py.
# Copyright (C) 2017-2019 The Software Heritage developers | # Copyright (C) 2017-2019 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 | ||||
import hashlib | import hashlib | ||||
from io import BytesIO | from io import BytesIO | ||||
from django.urls import reverse | from django.urls import reverse | ||||
from rest_framework import status | from rest_framework import status | ||||
from swh.deposit.config import ( | from swh.deposit.config import COL_IRI, DEPOSIT_STATUS_REJECTED | ||||
COL_IRI, | |||||
DEPOSIT_STATUS_LOAD_FAILURE, | |||||
DEPOSIT_STATUS_LOAD_SUCCESS, | |||||
DEPOSIT_STATUS_PARTIAL, | |||||
DEPOSIT_STATUS_REJECTED, | |||||
SE_IRI, | |||||
) | |||||
from swh.deposit.models import Deposit | |||||
from swh.deposit.parsers import parse_xml | from swh.deposit.parsers import parse_xml | ||||
from ..conftest import create_deposit | |||||
def test_deposit_post_will_fail_with_401(client): | def test_deposit_post_will_fail_with_401(client): | ||||
"""Without authentication, endpoint refuses access with 401 response | """Without authentication, endpoint refuses access with 401 response | ||||
""" | """ | ||||
url = reverse(COL_IRI, args=["hal"]) | url = reverse(COL_IRI, args=["hal"]) | ||||
response = client.post(url) | response = client.post(url) | ||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED | assert response.status_code == status.HTTP_401_UNAUTHORIZED | ||||
▲ Show 20 Lines • Show All 45 Lines • ▼ Show 20 Lines | response = authenticated_client.post( | ||||
HTTP_PACKAGING="http://purl.org/net/sword/package/SimpleZip", | HTTP_PACKAGING="http://purl.org/net/sword/package/SimpleZip", | ||||
HTTP_CONTENT_DISPOSITION="attachment; filename=filename0", | 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)) | ||||
actual_state = response_content["deposit_status"] | actual_state = response_content["deposit_status"] | ||||
assert actual_state == DEPOSIT_STATUS_REJECTED | assert actual_state == DEPOSIT_STATUS_REJECTED | ||||
def test_act_on_deposit_rejected_is_not_permitted( | |||||
authenticated_client, deposit_collection, rejected_deposit, atom_dataset | |||||
): | |||||
deposit = rejected_deposit | |||||
response = authenticated_client.post( | |||||
reverse(SE_IRI, args=[deposit.collection.name, deposit.id]), | |||||
content_type="application/atom+xml;type=entry", | |||||
data=atom_dataset["entry-data1"], | |||||
HTTP_SLUG=deposit.external_id, | |||||
) | |||||
assert response.status_code == status.HTTP_400_BAD_REQUEST | |||||
msg = "You can only act on deposit with status '%s'" % ( | |||||
DEPOSIT_STATUS_PARTIAL, | |||||
) | |||||
assert msg in response.content.decode("utf-8") | |||||
def test_add_deposit_when_partial_makes_new_deposit( | |||||
authenticated_client, | |||||
deposit_collection, | |||||
partial_deposit, | |||||
atom_dataset, | |||||
deposit_user, | |||||
): | |||||
"""Posting deposit on collection when previous is partial makes new deposit | |||||
""" | |||||
deposit = partial_deposit | |||||
assert deposit.status == DEPOSIT_STATUS_PARTIAL | |||||
origin_url = deposit_user.provider_url + deposit.external_id | |||||
# adding a new deposit with the same external id | |||||
response = authenticated_client.post( | |||||
reverse(COL_IRI, args=[deposit_collection.name]), | |||||
content_type="application/atom+xml;type=entry", | |||||
data=atom_dataset["entry-data0"] % origin_url, | |||||
HTTP_SLUG=deposit.external_id, | |||||
) | |||||
assert response.status_code == status.HTTP_201_CREATED, response.content.decode() | |||||
response_content = parse_xml(BytesIO(response.content)) | |||||
deposit_id = response_content["swh:deposit_id"] | |||||
assert deposit_id != deposit.id # new deposit | |||||
new_deposit = Deposit.objects.get(pk=deposit_id) | |||||
assert new_deposit != deposit | |||||
assert new_deposit.parent is None | |||||
def test_add_deposit_when_failed_makes_new_deposit_with_no_parent( | |||||
authenticated_client, deposit_collection, failed_deposit, atom_dataset, deposit_user | |||||
): | |||||
"""Posting deposit on collection when deposit done makes new deposit with | |||||
parent | |||||
""" | |||||
deposit = failed_deposit | |||||
assert deposit.status == DEPOSIT_STATUS_LOAD_FAILURE | |||||
origin_url = deposit_user.provider_url + deposit.external_id | |||||
# adding a new deposit with the same external id as a completed deposit | |||||
# creates the parenting chain | |||||
response = authenticated_client.post( | |||||
reverse(COL_IRI, args=[deposit_collection.name]), | |||||
content_type="application/atom+xml;type=entry", | |||||
data=atom_dataset["entry-data0"] % origin_url, | |||||
HTTP_SLUG=deposit.external_id, | |||||
) | |||||
assert response.status_code == status.HTTP_201_CREATED | |||||
response_content = parse_xml(BytesIO(response.content)) | |||||
deposit_id = response_content["swh:deposit_id"] | |||||
assert deposit_id != deposit.id | |||||
new_deposit = Deposit.objects.get(pk=deposit_id) | |||||
assert new_deposit != deposit | |||||
assert new_deposit.parent is None | |||||
def test_add_deposit_when_done_makes_new_deposit_with_parent_old_one( | |||||
authenticated_client, | |||||
deposit_collection, | |||||
completed_deposit, | |||||
atom_dataset, | |||||
deposit_user, | |||||
): | |||||
"""Posting deposit on collection when deposit done makes new deposit with | |||||
parent | |||||
""" | |||||
# given multiple deposit already loaded | |||||
deposit = completed_deposit | |||||
assert deposit.status == DEPOSIT_STATUS_LOAD_SUCCESS | |||||
origin_url = deposit_user.provider_url + deposit.external_id | |||||
# adding a new deposit with the same external id as a completed deposit | |||||
# creates the parenting chain | |||||
response = authenticated_client.post( | |||||
reverse(COL_IRI, args=[deposit_collection.name]), | |||||
content_type="application/atom+xml;type=entry", | |||||
data=atom_dataset["entry-data0"] % origin_url, | |||||
HTTP_SLUG=deposit.external_id, | |||||
) | |||||
assert response.status_code == status.HTTP_201_CREATED | |||||
response_content = parse_xml(BytesIO(response.content)) | |||||
deposit_id = response_content["swh:deposit_id"] | |||||
assert deposit_id != deposit.id | |||||
new_deposit = Deposit.objects.get(pk=deposit_id) | |||||
assert deposit.collection == new_deposit.collection | |||||
assert deposit.origin_url == origin_url | |||||
assert new_deposit != deposit | |||||
assert new_deposit.parent == deposit | |||||
def test_add_deposit_with_add_to_origin( | |||||
authenticated_client, | |||||
deposit_collection, | |||||
completed_deposit, | |||||
atom_dataset, | |||||
deposit_user, | |||||
): | |||||
"""Posting deposit with <swh:add_to_origin> creates a new deposit with parent | |||||
""" | |||||
# given multiple deposit already loaded | |||||
deposit = completed_deposit | |||||
assert deposit.status == DEPOSIT_STATUS_LOAD_SUCCESS | |||||
origin_url = deposit_user.provider_url + deposit.external_id | |||||
# adding a new deposit with the same external id as a completed deposit | |||||
# creates the parenting chain | |||||
response = authenticated_client.post( | |||||
reverse(COL_IRI, args=[deposit_collection.name]), | |||||
content_type="application/atom+xml;type=entry", | |||||
data=atom_dataset["entry-data-with-add-to-origin"] % origin_url, | |||||
) | |||||
assert response.status_code == status.HTTP_201_CREATED | |||||
response_content = parse_xml(BytesIO(response.content)) | |||||
deposit_id = response_content["swh:deposit_id"] | |||||
assert deposit_id != deposit.id | |||||
new_deposit = Deposit.objects.get(pk=deposit_id) | |||||
assert deposit.collection == new_deposit.collection | |||||
assert deposit.origin_url == origin_url | |||||
assert new_deposit != deposit | |||||
assert new_deposit.parent == deposit | |||||
def test_add_deposit_external_id_conflict_no_parent( | |||||
authenticated_client, | |||||
another_authenticated_client, | |||||
deposit_collection, | |||||
deposit_another_collection, | |||||
atom_dataset, | |||||
sample_archive, | |||||
deposit_user, | |||||
): | |||||
"""Posting a deposit with an external_id conflicting with an external_id | |||||
of a different client does not create a parent relationship | |||||
""" | |||||
external_id = "foobar" | |||||
origin_url = deposit_user.provider_url + external_id | |||||
# create a deposit for that other user, with the same slug | |||||
other_deposit = create_deposit( | |||||
another_authenticated_client, | |||||
deposit_another_collection.name, | |||||
sample_archive, | |||||
external_id, | |||||
DEPOSIT_STATUS_LOAD_SUCCESS, | |||||
) | |||||
# adding a new deposit with the same external id as a completed deposit | |||||
response = authenticated_client.post( | |||||
reverse(COL_IRI, args=[deposit_collection.name]), | |||||
content_type="application/atom+xml;type=entry", | |||||
data=atom_dataset["entry-data0"] % origin_url, | |||||
HTTP_SLUG=external_id, | |||||
) | |||||
assert response.status_code == status.HTTP_201_CREATED | |||||
response_content = parse_xml(BytesIO(response.content)) | |||||
deposit_id = response_content["swh:deposit_id"] | |||||
assert other_deposit.id != deposit_id | |||||
new_deposit = Deposit.objects.get(pk=deposit_id) | |||||
assert new_deposit.parent is None | |||||
def test_add_deposit_external_id_conflict_with_parent( | |||||
authenticated_client, | |||||
another_authenticated_client, | |||||
deposit_collection, | |||||
deposit_another_collection, | |||||
completed_deposit, | |||||
atom_dataset, | |||||
sample_archive, | |||||
deposit_user, | |||||
): | |||||
"""Posting a deposit with an external_id conflicting with an external_id | |||||
of a different client creates a parent relationship with the deposit | |||||
of the right client instead of the last matching deposit | |||||
This test does not have an equivalent for origin url conflicts, as these | |||||
can not happen (assuming clients do not have provider_url overlaps) | |||||
""" | |||||
# given multiple deposit already loaded | |||||
deposit = completed_deposit | |||||
assert deposit.status == DEPOSIT_STATUS_LOAD_SUCCESS | |||||
origin_url = deposit_user.provider_url + deposit.external_id | |||||
# create a deposit for that other user, with the same slug | |||||
other_deposit = create_deposit( | |||||
another_authenticated_client, | |||||
deposit_another_collection.name, | |||||
sample_archive, | |||||
deposit.external_id, | |||||
DEPOSIT_STATUS_LOAD_SUCCESS, | |||||
) | |||||
# adding a new deposit with the same external id as a completed deposit | |||||
response = authenticated_client.post( | |||||
reverse(COL_IRI, args=[deposit_collection.name]), | |||||
content_type="application/atom+xml;type=entry", | |||||
data=atom_dataset["entry-data0"] % origin_url, | |||||
HTTP_SLUG=deposit.external_id, | |||||
) | |||||
assert response.status_code == status.HTTP_201_CREATED | |||||
response_content = parse_xml(BytesIO(response.content)) | |||||
deposit_id = response_content["swh:deposit_id"] | |||||
assert deposit_id != deposit.id | |||||
assert other_deposit.id != deposit.id | |||||
new_deposit = Deposit.objects.get(pk=deposit_id) | |||||
assert deposit.collection == new_deposit.collection | |||||
assert deposit.external_id == new_deposit.external_id | |||||
assert new_deposit != deposit | |||||
assert new_deposit.parent == deposit | |||||
def test_add_deposit_add_to_origin_conflict( | |||||
authenticated_client, | |||||
another_authenticated_client, | |||||
deposit_collection, | |||||
deposit_another_collection, | |||||
atom_dataset, | |||||
sample_archive, | |||||
deposit_user, | |||||
deposit_another_user, | |||||
): | |||||
"""Posting a deposit with an <swh:add_to_origin> referencing an origin | |||||
owned by a different client raises an error | |||||
""" | |||||
external_id = "foobar" | |||||
origin_url = deposit_another_user.provider_url + external_id | |||||
# create a deposit for that other user, with the same slug | |||||
create_deposit( | |||||
another_authenticated_client, | |||||
deposit_another_collection.name, | |||||
sample_archive, | |||||
external_id, | |||||
DEPOSIT_STATUS_LOAD_SUCCESS, | |||||
) | |||||
# adding a new deposit with the same external id as a completed deposit | |||||
response = authenticated_client.post( | |||||
reverse(COL_IRI, args=[deposit_collection.name]), | |||||
content_type="application/atom+xml;type=entry", | |||||
data=atom_dataset["entry-data0"] % origin_url, | |||||
) | |||||
assert response.status_code == status.HTTP_403_FORBIDDEN | |||||
assert b"must start with" in response.content | |||||
def test_add_deposit_add_to_wrong_origin( | |||||
authenticated_client, deposit_collection, atom_dataset, sample_archive, | |||||
): | |||||
"""Posting a deposit with an <swh:add_to_origin> referencing an origin | |||||
not starting with the provider_url raises an error | |||||
""" | |||||
origin_url = "http://example.org/foo" | |||||
# adding a new deposit with the same external id as a completed deposit | |||||
response = authenticated_client.post( | |||||
reverse(COL_IRI, args=[deposit_collection.name]), | |||||
content_type="application/atom+xml;type=entry", | |||||
data=atom_dataset["entry-data0"] % origin_url, | |||||
) | |||||
assert response.status_code == status.HTTP_403_FORBIDDEN | |||||
assert b"must start with" in response.content |