diff --git a/swh/deposit/tests/api/data/atom/entry-data-deposit-binary.xml b/swh/deposit/tests/api/data/atom/entry-data-deposit-binary.xml new file mode 100644 index 00000000..65b7f63b --- /dev/null +++ b/swh/deposit/tests/api/data/atom/entry-data-deposit-binary.xml @@ -0,0 +1,29 @@ + + + Title + urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a + 2005-10-07T17:17:08Z + Contributor + The abstract + + + The abstract + Access Rights + Alternative Title + Date Available + Bibliographic Citation # noqa + Contributor + Description + Has Part + Has Version + Identifier + Is Part Of + Publisher + References + Rights Holder + Source + Title + Type + + diff --git a/swh/deposit/tests/api/test_deposit_binary.py b/swh/deposit/tests/api/test_deposit_binary.py index 59aeb15c..d4f55c07 100644 --- a/swh/deposit/tests/api/test_deposit_binary.py +++ b/swh/deposit/tests/api/test_deposit_binary.py @@ -1,645 +1,580 @@ # Copyright (C) 2017-2019 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information +import pytest +import re + from django.core.files.uploadedfile import InMemoryUploadedFile from django.urls import reverse from io import BytesIO from rest_framework import status -from rest_framework.test import APITestCase from swh.deposit.tests import TEST_CONFIG -from swh.deposit.config import COL_IRI, EM_IRI -from swh.deposit.config import DEPOSIT_STATUS_DEPOSITED +from swh.deposit.config import ( + COL_IRI, EM_IRI, DEPOSIT_STATUS_DEPOSITED, +) from swh.deposit.models import Deposit, DepositRequest from swh.deposit.parsers import parse_xml -from ..common import ( - BasicTestCase, WithAuthTestCase, create_arborescence_archive, - FileSystemCreationRoutine -) +from swh.deposit.tests.common import create_arborescence_archive + + +def check_archive(archive_name: str, archive_name_to_check: str): + """Helper function to ensure archive_name is present within the + archive_name_to_check. + + Raises: + AssertionError if archive_name is not present within + archive_name_to_check + + """ + if '.' in archive_name: + filename, extension = archive_name.split('.') + pattern = re.compile('.*/%s.*\\.%s' % (filename, extension)) + else: + pattern = re.compile('.*/%s' % archive_name) + assert pattern.match(archive_name_to_check) is not None + + +def test_check_archive_helper(): + # success + for archive_name, archive_name_to_check in [ + ('filename0', 'something/filename0'), + ('archive.zip', 'client_1/archive_noisynoise.zip'), + ]: + check_archive(archive_name, archive_name_to_check) + + # failures + for archive_name, archive_name_to_check in [ + ('filename0', 'something-filename0'), + ('archive.zip', 'client_1_archive_noisynoise.zip'), + ('reference', 'irrelevant'), + ]: + with pytest.raises(AssertionError): + check_archive(archive_name, archive_name_to_check) + + +def test_post_deposit_binary_no_slug( + authenticated_client, deposit_collection, sample_archive): + """Posting a binary deposit without slug header should return 400 + + """ + url = reverse(COL_IRI, args=[deposit_collection.name]) + + # when + response = authenticated_client.post( + url, + 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 b'Missing SLUG header' in response.content + assert response.status_code == status.HTTP_400_BAD_REQUEST + + +def test_post_deposit_binary_support( + authenticated_client, deposit_collection, sample_archive): + """Binary upload with content-type not in [zip,x-tar] should return 415 + + """ + # given + url = reverse(COL_IRI, args=[deposit_collection.name]) + + external_id = 'some-external-id-1' + + # when + response = authenticated_client.post( + url, + content_type='application/octet-stream', + data=sample_archive['data'], + # + headers + CONTENT_LENGTH=sample_archive['length'], + HTTP_SLUG=external_id, + 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') + + # then + assert response.status_code == status.HTTP_415_UNSUPPORTED_MEDIA_TYPE + + with pytest.raises(Deposit.DoesNotExist): + Deposit.objects.get(external_id=external_id) + + +def test_post_deposit_binary_upload_ok( + authenticated_client, deposit_collection, sample_archive): + """Binary upload with correct headers should return 201 with receipt + + """ + # given + url = reverse(COL_IRI, args=[deposit_collection.name]) + + external_id = 'some-external-id-1' + + # when + response = authenticated_client.post( + url, + content_type='application/zip', # as zip + 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_CONTENT_MD5=sample_archive['md5sum'], + HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', + HTTP_IN_PROGRESS='false', + HTTP_CONTENT_DISPOSITION='attachment; filename=%s' % ( + sample_archive['name'], )) + + # then + response_content = parse_xml(BytesIO(response.content)) + assert response.status_code == status.HTTP_201_CREATED + deposit_id = response_content['deposit_id'] + + deposit = Deposit.objects.get(pk=deposit_id) + assert deposit.status == DEPOSIT_STATUS_DEPOSITED + assert deposit.external_id == external_id + assert deposit.collection == deposit_collection + assert deposit.swh_id is None + + deposit_request = DepositRequest.objects.get(deposit=deposit) + check_archive(sample_archive['name'], deposit_request.archive.name) + + assert deposit_request.metadata is None + assert deposit_request.raw_metadata is None + + response_content = parse_xml(BytesIO(response.content)) + assert response_content['deposit_archive'] == sample_archive['name'] + assert int(response_content['deposit_id']) == deposit.id + assert response_content['deposit_status'] == deposit.status + + edit_se_iri = reverse('edit_se_iri', + args=[deposit_collection.name, deposit.id]) + + assert response._headers['location'] == ( + 'Location', 'http://testserver' + edit_se_iri) + + +def test_post_deposit_binary_failure_unsupported_packaging_header( + authenticated_client, deposit_collection, sample_archive): + """Bin deposit without supported content_disposition header returns 400 + + """ + # given + url = reverse(COL_IRI, args=[deposit_collection.name]) + + external_id = 'some-external-id' + + # when + response = authenticated_client.post( + url, + content_type='application/zip', + data=sample_archive['data'], + # + headers + CONTENT_LENGTH=sample_archive['length'], + HTTP_SLUG=external_id, + HTTP_CONTENT_MD5=sample_archive['md5sum'], + HTTP_PACKAGING='something-unsupported', + HTTP_CONTENT_DISPOSITION='attachment; filename=filename0') + + # then + assert response.status_code == status.HTTP_400_BAD_REQUEST + with pytest.raises(Deposit.DoesNotExist): + Deposit.objects.get(external_id=external_id) + + +def test_post_deposit_binary_upload_no_content_disposition_header( + authenticated_client, deposit_collection, sample_archive): + """Binary upload without content_disposition header should return 400 + + """ + # given + url = reverse(COL_IRI, args=[deposit_collection.name]) + + external_id = 'some-external-id' + + # when + response = authenticated_client.post( + url, + content_type='application/zip', + data=sample_archive['data'], + # + headers + CONTENT_LENGTH=sample_archive['length'], + HTTP_SLUG=external_id, + HTTP_CONTENT_MD5=sample_archive['md5sum'], + HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', + HTTP_IN_PROGRESS='false') + # then + assert response.status_code == status.HTTP_400_BAD_REQUEST + with pytest.raises(Deposit.DoesNotExist): + Deposit.objects.get(external_id=external_id) -class DepositTestCase(APITestCase, WithAuthTestCase, BasicTestCase, - FileSystemCreationRoutine): - """Try and upload one single deposit + +def test_post_deposit_mediation_not_supported( + authenticated_client, deposit_collection, sample_archive): + """Binary upload with mediation should return a 412 response + + """ + # given + url = reverse(COL_IRI, args=[deposit_collection.name]) + + external_id = 'some-external-id-1' + + # when + response = authenticated_client.post( + url, + content_type='application/zip', + data=sample_archive['data'], + # + headers + CONTENT_LENGTH=sample_archive['length'], + HTTP_SLUG=external_id, + HTTP_CONTENT_MD5=sample_archive['md5sum'], + HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', + HTTP_IN_PROGRESS='false', + HTTP_ON_BEHALF_OF='someone', + HTTP_CONTENT_DISPOSITION='attachment; filename=filename0') + + # then + assert response.status_code == status.HTTP_412_PRECONDITION_FAILED + + with pytest.raises(Deposit.DoesNotExist): + Deposit.objects.get(external_id=external_id) + + +def test_post_deposit_binary_upload_fail_if_upload_size_limit_exceeded( + authenticated_client, deposit_collection, sample_archive, tmp_path): + """Binary upload must not exceed the limit set up... + + """ + tmp_path = str(tmp_path) + url = reverse(COL_IRI, args=[deposit_collection.name]) + + archive = create_arborescence_archive( + tmp_path, 'archive2', 'file2', b'some content in file', + up_to_size=TEST_CONFIG['max_upload_size']) + + external_id = 'some-external-id' + + # when + response = authenticated_client.post( + url, + content_type='application/zip', + data=archive['data'], + # + headers + CONTENT_LENGTH=archive['length'], + HTTP_SLUG=external_id, + HTTP_CONTENT_MD5=archive['md5sum'], + HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', + HTTP_IN_PROGRESS='false', + HTTP_CONTENT_DISPOSITION='attachment; filename=filename0') + + # then + assert response.status_code == status.HTTP_413_REQUEST_ENTITY_TOO_LARGE + assert b'Upload size limit exceeded' in response.content + + with pytest.raises(Deposit.DoesNotExist): + Deposit.objects.get(external_id=external_id) + + +def test_post_deposit_2_post_2_different_deposits( + authenticated_client, deposit_collection, sample_archive): + """2 posting deposits should return 2 different 201 with receipt + + """ + url = reverse(COL_IRI, args=[deposit_collection.name]) + + # when + response = authenticated_client.post( + url, + content_type='application/zip', # as zip + data=sample_archive['data'], + # + headers + CONTENT_LENGTH=sample_archive['length'], + 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_CONTENT_DISPOSITION='attachment; filename=filename0') + + # then + assert response.status_code == status.HTTP_201_CREATED + + response_content = parse_xml(BytesIO(response.content)) + deposit_id = response_content['deposit_id'] + + deposit = Deposit.objects.get(pk=deposit_id) + + deposits = Deposit.objects.all() + assert len(deposits) == 1 + assert deposits[0] == deposit + + # second post + response = authenticated_client.post( + url, + content_type='application/x-tar', # as zip + data=sample_archive['data'], + # + headers + CONTENT_LENGTH=sample_archive['length'], + 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_CONTENT_DISPOSITION='attachment; filename=filename1') + + assert response.status_code == status.HTTP_201_CREATED + + response_content = parse_xml(BytesIO(response.content)) + deposit_id2 = response_content['deposit_id'] + + deposit2 = Deposit.objects.get(pk=deposit_id2) + + assert deposit != deposit2 + + deposits = Deposit.objects.all().order_by('id') + assert len(deposits) == 2 + assert list(deposits), [deposit == deposit2] + + +def test_post_deposit_binary_and_post_to_add_another_archive( + authenticated_client, deposit_collection, sample_archive, tmp_path): + """Updating a deposit should return a 201 with receipt + + """ + tmp_path = str(tmp_path) + url = reverse(COL_IRI, args=[deposit_collection.name]) + + external_id = 'some-external-id-1' + + # when + response = authenticated_client.post( + url, + content_type='application/zip', # as zip + data=sample_archive['data'], + # + headers + CONTENT_LENGTH=sample_archive['length'], + HTTP_SLUG=external_id, + HTTP_CONTENT_MD5=sample_archive['md5sum'], + HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', + HTTP_IN_PROGRESS='true', + HTTP_CONTENT_DISPOSITION='attachment; filename=%s' % ( + sample_archive['name'], )) + + # then + assert response.status_code == status.HTTP_201_CREATED + + response_content = parse_xml(BytesIO(response.content)) + deposit_id = response_content['deposit_id'] + + deposit = Deposit.objects.get(pk=deposit_id) + assert deposit.status == 'partial' + assert deposit.external_id == external_id + assert deposit.collection == deposit_collection + assert deposit.swh_id is None + + deposit_request = DepositRequest.objects.get(deposit=deposit) + assert deposit_request.deposit == deposit + assert deposit_request.type == 'archive' + check_archive(sample_archive['name'], deposit_request.archive.name) + + # 2nd archive to upload + archive2 = create_arborescence_archive( + tmp_path, 'archive2', 'file2', b'some other content in file') + + # uri to update the content + update_uri = reverse(EM_IRI, args=[deposit_collection.name, deposit_id]) + + # adding another archive for the deposit and finalizing it + response = authenticated_client.post( + update_uri, + content_type='application/zip', # as zip + data=archive2['data'], + # + headers + CONTENT_LENGTH=archive2['length'], + HTTP_SLUG=external_id, + HTTP_CONTENT_MD5=archive2['md5sum'], + HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', + HTTP_CONTENT_DISPOSITION='attachment; filename=%s' % ( + archive2['name'])) + + assert response.status_code == status.HTTP_201_CREATED + response_content = parse_xml(BytesIO(response.content)) + + deposit = Deposit.objects.get(pk=deposit_id) + assert deposit.status == DEPOSIT_STATUS_DEPOSITED + assert deposit.external_id == external_id + assert deposit.collection == deposit_collection + assert deposit.swh_id is None + + deposit_requests = list(DepositRequest.objects.filter(deposit=deposit). + order_by('id')) + + # 2 deposit requests for the same deposit + assert len(deposit_requests) == 2 + assert deposit_requests[0].deposit == deposit + assert deposit_requests[0].type == 'archive' + check_archive(sample_archive['name'], deposit_requests[0].archive.name) + + assert deposit_requests[1].deposit == deposit + assert deposit_requests[1].type == 'archive' + check_archive(archive2['name'], deposit_requests[1].archive.name) + + # only 1 deposit in db + deposits = Deposit.objects.all() + assert len(deposits) == 1 + + +def test_post_deposit_then_update_refused( + authenticated_client, deposit_collection, + sample_archive, atom_dataset, tmp_path): + """Updating a deposit with status 'ready' should return a 400 """ - def setUp(self): - super().setUp() - - self.atom_entry_data0 = b""" - - Awesome Compiler - hal - urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a - %s - 2017-10-07T15:17:08Z - some awesome author - something - awesome-compiler - This is an awesome compiler destined to -awesomely compile stuff -and other stuff - compiler,programming,language - 2005-10-07T17:17:08Z - 2005-10-07T17:17:08Z - release note - related link - - Awesome - https://hoster.org/awesome-compiler - GNU/Linux - 0.0.1 - running - all -""" - - self.atom_entry_data1 = b""" - - hal - urn:uuid:2225c695-cfb8-4ebb-aaaa-80da344efa6a - 2017-10-07T15:17:08Z - some awesome author - something - awesome-compiler - This is an awesome compiler destined to -awesomely compile stuff -and other stuff - compiler,programming,language - 2005-10-07T17:17:08Z - 2005-10-07T17:17:08Z - release note - related link - - Awesome - https://hoster.org/awesome-compiler - GNU/Linux - 0.0.1 - running - all -""" - - self.atom_entry_data2 = b""" - - %s -""" - - self.atom_entry_data_empty_body = b""" -""" - - self.atom_entry_data3 = b""" - - something -""" - - self.data_atom_entry_ok = b""" - - Title - urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a - 2005-10-07T17:17:08Z - Contributor - The abstract - - - The abstract - Access Rights - Alternative Title - Date Available - Bibliographic Citation # noqa - Contributor - Description - Has Part - Has Version - Identifier - Is Part Of - Publisher - References - Rights Holder - Source - Title - Type - -""" - - def test_post_deposit_binary_without_slug_header_is_bad_request(self): - """Posting a binary deposit without slug header should return 400 - - """ - url = reverse(COL_IRI, args=[self.collection.name]) - - # when - response = self.client.post( - url, - content_type='application/zip', # as zip - data=self.archive['data'], - # + headers - CONTENT_LENGTH=self.archive['length'], - HTTP_CONTENT_MD5=self.archive['md5sum'], - HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', - HTTP_IN_PROGRESS='false', - HTTP_CONTENT_DISPOSITION='attachment; filename=filename0') - - self.assertIn(b'Missing SLUG header', response.content) - self.assertEqual(response.status_code, - status.HTTP_400_BAD_REQUEST) - - def test_post_deposit_binary_upload_final_and_status_check(self): - """Binary upload with correct headers should return 201 with receipt - - """ - # given - url = reverse(COL_IRI, args=[self.collection.name]) - - external_id = 'some-external-id-1' - - # when - response = self.client.post( - url, - content_type='application/zip', # as zip - data=self.archive['data'], - # + headers - CONTENT_LENGTH=self.archive['length'], - # other headers needs HTTP_ prefix to be taken into account - HTTP_SLUG=external_id, - HTTP_CONTENT_MD5=self.archive['md5sum'], - HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', - HTTP_IN_PROGRESS='false', - HTTP_CONTENT_DISPOSITION='attachment; filename=%s' % ( - self.archive['name'], )) - - # then - response_content = parse_xml(BytesIO(response.content)) - self.assertEqual(response.status_code, status.HTTP_201_CREATED) - deposit_id = response_content['deposit_id'] - - deposit = Deposit.objects.get(pk=deposit_id) - self.assertEqual(deposit.status, DEPOSIT_STATUS_DEPOSITED) - self.assertEqual(deposit.external_id, external_id) - self.assertEqual(deposit.collection, self.collection) - self.assertEqual(deposit.client, self.user) - self.assertIsNone(deposit.swh_id) - - deposit_request = DepositRequest.objects.get(deposit=deposit) - self.assertEqual(deposit_request.deposit, deposit) - self.assertRegex(deposit_request.archive.name, self.archive['name']) - self.assertIsNone(deposit_request.metadata) - self.assertIsNone(deposit_request.raw_metadata) - - response_content = parse_xml(BytesIO(response.content)) - self.assertEqual(response_content['deposit_archive'], - self.archive['name']) - self.assertEqual(int(response_content['deposit_id']), - deposit.id) - self.assertEqual(response_content['deposit_status'], - deposit.status) - - edit_se_iri = reverse('edit_se_iri', - args=[self.collection.name, deposit.id]) - - self.assertEqual(response._headers['location'], - ('Location', 'http://testserver' + edit_se_iri)) - - def test_post_deposit_binary_upload_supports_zip_or_tar(self): - """Binary upload with content-type not in [zip,x-tar] should return 415 - - """ - # given - url = reverse(COL_IRI, args=[self.collection.name]) - - external_id = 'some-external-id-1' - - # when - response = self.client.post( - url, - content_type='application/octet-stream', - data=self.archive['data'], - # + headers - CONTENT_LENGTH=self.archive['length'], - HTTP_SLUG=external_id, - HTTP_CONTENT_MD5=self.archive['md5sum'], - HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', - HTTP_IN_PROGRESS='false', - HTTP_CONTENT_DISPOSITION='attachment; filename=filename0') - - # then - self.assertEqual(response.status_code, - status.HTTP_415_UNSUPPORTED_MEDIA_TYPE) - - with self.assertRaises(Deposit.DoesNotExist): - Deposit.objects.get(external_id=external_id) - - def test_post_deposit_binary_fails_if_unsupported_packaging_header( - self): - """Bin deposit without supported content_disposition header returns 400 - - """ - # given - url = reverse(COL_IRI, args=[self.collection.name]) - - external_id = 'some-external-id' - - # when - response = self.client.post( - url, - content_type='application/zip', - data=self.archive['data'], - # + headers - CONTENT_LENGTH=self.archive['length'], - HTTP_SLUG=external_id, - HTTP_CONTENT_MD5=self.archive['md5sum'], - HTTP_PACKAGING='something-unsupported', - HTTP_CONTENT_DISPOSITION='attachment; filename=filename0') - - # then - self.assertEqual(response.status_code, - status.HTTP_400_BAD_REQUEST) - with self.assertRaises(Deposit.DoesNotExist): - Deposit.objects.get(external_id=external_id) - - def test_post_deposit_binary_upload_fail_if_no_content_disposition_header( - self): - """Binary upload without content_disposition header should return 400 - - """ - # given - url = reverse(COL_IRI, args=[self.collection.name]) - - external_id = 'some-external-id' - - # when - response = self.client.post( - url, - content_type='application/zip', - data=self.archive['data'], - # + headers - CONTENT_LENGTH=self.archive['length'], - HTTP_SLUG=external_id, - HTTP_CONTENT_MD5=self.archive['md5sum'], - HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', - HTTP_IN_PROGRESS='false') - - # then - self.assertEqual(response.status_code, - status.HTTP_400_BAD_REQUEST) - with self.assertRaises(Deposit.DoesNotExist): - Deposit.objects.get(external_id=external_id) - - def test_post_deposit_mediation_not_supported(self): - """Binary upload with mediation should return a 412 response - - """ - # given - url = reverse(COL_IRI, args=[self.collection.name]) - - external_id = 'some-external-id-1' - - # when - response = self.client.post( - url, - content_type='application/zip', - data=self.archive['data'], - # + headers - CONTENT_LENGTH=self.archive['length'], - HTTP_SLUG=external_id, - HTTP_CONTENT_MD5=self.archive['md5sum'], - HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', - HTTP_IN_PROGRESS='false', - HTTP_ON_BEHALF_OF='someone', - HTTP_CONTENT_DISPOSITION='attachment; filename=filename0') - - # then - self.assertEqual(response.status_code, - status.HTTP_412_PRECONDITION_FAILED) - - with self.assertRaises(Deposit.DoesNotExist): - Deposit.objects.get(external_id=external_id) - - def test_post_deposit_binary_upload_fail_if_upload_size_limit_exceeded( - self): - """Binary upload must not exceed the limit set up... - - """ - # given - url = reverse(COL_IRI, args=[self.collection.name]) - - archive = create_arborescence_archive( - self.root_path, 'archive2', 'file2', b'some content in file', - up_to_size=TEST_CONFIG['max_upload_size']) - - external_id = 'some-external-id' - - # when - response = self.client.post( - url, - content_type='application/zip', - data=archive['data'], - # + headers - CONTENT_LENGTH=archive['length'], - HTTP_SLUG=external_id, - HTTP_CONTENT_MD5=archive['md5sum'], - HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', - HTTP_IN_PROGRESS='false', - HTTP_CONTENT_DISPOSITION='attachment; filename=filename0') - - # then - self.assertEqual(response.status_code, - status.HTTP_413_REQUEST_ENTITY_TOO_LARGE) - self.assertRegex(response.content, b'Upload size limit exceeded') - - with self.assertRaises(Deposit.DoesNotExist): - Deposit.objects.get(external_id=external_id) - - def test_post_deposit_2_post_2_different_deposits(self): - """2 posting deposits should return 2 different 201 with receipt - - """ - url = reverse(COL_IRI, args=[self.collection.name]) - - # when - response = self.client.post( - url, - content_type='application/zip', # as zip - data=self.archive['data'], - # + headers - CONTENT_LENGTH=self.archive['length'], - HTTP_SLUG='some-external-id-1', - HTTP_CONTENT_MD5=self.archive['md5sum'], - HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', - HTTP_IN_PROGRESS='false', - HTTP_CONTENT_DISPOSITION='attachment; filename=filename0') - - # then - self.assertEqual(response.status_code, status.HTTP_201_CREATED) - - response_content = parse_xml(BytesIO(response.content)) - deposit_id = response_content['deposit_id'] - - deposit = Deposit.objects.get(pk=deposit_id) - - deposits = Deposit.objects.all() - self.assertEqual(len(deposits), 1) - self.assertEqual(deposits[0], deposit) - - # second post - response = self.client.post( - url, - content_type='application/x-tar', # as zip - data=self.archive['data'], - # + headers - CONTENT_LENGTH=self.archive['length'], - HTTP_SLUG='another-external-id', - HTTP_CONTENT_MD5=self.archive['md5sum'], - HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', - HTTP_IN_PROGRESS='false', - HTTP_CONTENT_DISPOSITION='attachment; filename=filename1') - - self.assertEqual(response.status_code, status.HTTP_201_CREATED) - - response_content = parse_xml(BytesIO(response.content)) - deposit_id2 = response_content['deposit_id'] - - deposit2 = Deposit.objects.get(pk=deposit_id2) - - self.assertNotEqual(deposit, deposit2) - - deposits = Deposit.objects.all().order_by('id') - self.assertEqual(len(deposits), 2) - self.assertEqual(list(deposits), [deposit, deposit2]) - - def test_post_deposit_binary_and_post_to_add_another_archive(self): - """Updating a deposit should return a 201 with receipt - - """ - # given - url = reverse(COL_IRI, args=[self.collection.name]) - - external_id = 'some-external-id-1' - - # when - response = self.client.post( - url, - content_type='application/zip', # as zip - data=self.archive['data'], - # + headers - CONTENT_LENGTH=self.archive['length'], - HTTP_SLUG=external_id, - HTTP_CONTENT_MD5=self.archive['md5sum'], - HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', - HTTP_IN_PROGRESS='true', - HTTP_CONTENT_DISPOSITION='attachment; filename=%s' % ( - self.archive['name'], )) - - # then - self.assertEqual(response.status_code, status.HTTP_201_CREATED) - - response_content = parse_xml(BytesIO(response.content)) - deposit_id = response_content['deposit_id'] - - deposit = Deposit.objects.get(pk=deposit_id) - self.assertEqual(deposit.status, 'partial') - self.assertEqual(deposit.external_id, external_id) - self.assertEqual(deposit.collection, self.collection) - self.assertEqual(deposit.client, self.user) - self.assertIsNone(deposit.swh_id) - - deposit_request = DepositRequest.objects.get(deposit=deposit) - self.assertEqual(deposit_request.deposit, deposit) - self.assertEqual(deposit_request.type, 'archive') - self.assertRegex(deposit_request.archive.name, self.archive['name']) - - # 2nd archive to upload - archive2 = create_arborescence_archive( - self.root_path, 'archive2', 'file2', b'some other content in file') - - # uri to update the content - update_uri = reverse(EM_IRI, args=[self.collection.name, deposit_id]) - - # adding another archive for the deposit and finalizing it - response = self.client.post( - update_uri, - content_type='application/zip', # as zip - data=archive2['data'], - # + headers - CONTENT_LENGTH=archive2['length'], - HTTP_SLUG=external_id, - HTTP_CONTENT_MD5=archive2['md5sum'], - HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', - HTTP_CONTENT_DISPOSITION='attachment; filename=%s' % ( - archive2['name'])) - - self.assertEqual(response.status_code, status.HTTP_201_CREATED) - response_content = parse_xml(BytesIO(response.content)) - - deposit = Deposit.objects.get(pk=deposit_id) - self.assertEqual(deposit.status, DEPOSIT_STATUS_DEPOSITED) - self.assertEqual(deposit.external_id, external_id) - self.assertEqual(deposit.collection, self.collection) - self.assertEqual(deposit.client, self.user) - self.assertIsNone(deposit.swh_id) - - deposit_requests = list(DepositRequest.objects.filter(deposit=deposit). - order_by('id')) - - # 2 deposit requests for the same deposit - self.assertEqual(len(deposit_requests), 2) - self.assertEqual(deposit_requests[0].deposit, deposit) - self.assertEqual(deposit_requests[0].type, 'archive') - self.assertRegex(deposit_requests[0].archive.name, - self.archive['name']) - - self.assertEqual(deposit_requests[1].deposit, deposit) - self.assertEqual(deposit_requests[1].type, 'archive') - self.assertRegex(deposit_requests[1].archive.name, - archive2['name']) - - # only 1 deposit in db - deposits = Deposit.objects.all() - self.assertEqual(len(deposits), 1) - - def test_post_deposit_then_post_or_put_is_refused_when_status_ready(self): - """Updating a deposit with status 'ready' should return a 400 - - """ - url = reverse(COL_IRI, args=[self.collection.name]) - - external_id = 'some-external-id-1' - - # when - response = self.client.post( - url, - content_type='application/zip', # as zip - data=self.archive['data'], - # + headers - CONTENT_LENGTH=self.archive['length'], - HTTP_SLUG=external_id, - HTTP_CONTENT_MD5=self.archive['md5sum'], - HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', - HTTP_IN_PROGRESS='false', - HTTP_CONTENT_DISPOSITION='attachment; filename=filename0') - - # then - self.assertEqual(response.status_code, status.HTTP_201_CREATED) - - response_content = parse_xml(BytesIO(response.content)) - deposit_id = response_content['deposit_id'] - - deposit = Deposit.objects.get(pk=deposit_id) - self.assertEqual(deposit.status, DEPOSIT_STATUS_DEPOSITED) - self.assertEqual(deposit.external_id, external_id) - self.assertEqual(deposit.collection, self.collection) - self.assertEqual(deposit.client, self.user) - self.assertIsNone(deposit.swh_id) - - deposit_request = DepositRequest.objects.get(deposit=deposit) - self.assertEqual(deposit_request.deposit, deposit) - self.assertRegex(deposit_request.archive.name, 'filename0') - - # updating/adding is forbidden - - # uri to update the content - edit_se_iri = reverse( - 'edit_se_iri', args=[self.collection.name, deposit_id]) - em_iri = reverse( - 'em_iri', args=[self.collection.name, deposit_id]) - - # Testing all update/add endpoint should fail - # since the status is ready - - archive2 = create_arborescence_archive( - self.root_path, 'archive2', 'file2', b'some content in file 2') - - # replacing file is no longer possible since the deposit's - # status is ready - r = self.client.put( - em_iri, - content_type='application/zip', - data=archive2['data'], - CONTENT_LENGTH=archive2['length'], - HTTP_SLUG=external_id, - HTTP_CONTENT_MD5=archive2['md5sum'], - HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', - HTTP_IN_PROGRESS='false', - HTTP_CONTENT_DISPOSITION='attachment; filename=filename0') - - self.assertEqual(r.status_code, status.HTTP_400_BAD_REQUEST) - - # adding file is no longer possible since the deposit's status - # is ready - r = self.client.post( - em_iri, - content_type='application/zip', - data=archive2['data'], - CONTENT_LENGTH=archive2['length'], - HTTP_SLUG=external_id, - HTTP_CONTENT_MD5=archive2['md5sum'], - HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', - HTTP_IN_PROGRESS='false', - HTTP_CONTENT_DISPOSITION='attachment; filename=filename0') - - self.assertEqual(r.status_code, status.HTTP_400_BAD_REQUEST) - - # replacing metadata is no longer possible since the deposit's - # status is ready - r = self.client.put( - edit_se_iri, - content_type='application/atom+xml;type=entry', - data=self.data_atom_entry_ok, - CONTENT_LENGTH=len(self.data_atom_entry_ok), - HTTP_SLUG=external_id) - - self.assertEqual(r.status_code, status.HTTP_400_BAD_REQUEST) - - # adding new metadata is no longer possible since the - # deposit's status is ready - r = self.client.post( - edit_se_iri, - content_type='application/atom+xml;type=entry', - data=self.data_atom_entry_ok, - CONTENT_LENGTH=len(self.data_atom_entry_ok), - HTTP_SLUG=external_id) - - self.assertEqual(r.status_code, status.HTTP_400_BAD_REQUEST) - - 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) - - atom_entry = InMemoryUploadedFile( - BytesIO(self.data_atom_entry_ok), - field_name='atom0', - name='atom0', - content_type='application/atom+xml; charset="utf-8"', - size=len(self.data_atom_entry_ok), - charset='utf-8') - - # replacing multipart metadata is no longer possible since the - # deposit's status is ready - r = self.client.put( - edit_se_iri, - format='multipart', - data={ - 'archive': archive, - 'atom_entry': atom_entry, - }) - - self.assertEqual(r.status_code, status.HTTP_400_BAD_REQUEST) - - # adding new metadata is no longer possible since the - # deposit's status is ready - r = self.client.post( - edit_se_iri, - format='multipart', - data={ - 'archive': archive, - 'atom_entry': atom_entry, - }) - - self.assertEqual(r.status_code, status.HTTP_400_BAD_REQUEST) + tmp_path = str(tmp_path) + url = reverse(COL_IRI, args=[deposit_collection.name]) + + external_id = 'some-external-id-1' + + # when + response = authenticated_client.post( + url, + content_type='application/zip', # as zip + data=sample_archive['data'], + # + headers + CONTENT_LENGTH=sample_archive['length'], + HTTP_SLUG=external_id, + 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') + + # then + assert response.status_code == status.HTTP_201_CREATED + + response_content = parse_xml(BytesIO(response.content)) + deposit_id = response_content['deposit_id'] + + deposit = Deposit.objects.get(pk=deposit_id) + assert deposit.status == DEPOSIT_STATUS_DEPOSITED + assert deposit.external_id == external_id + assert deposit.collection == deposit_collection + assert deposit.swh_id is None + + deposit_request = DepositRequest.objects.get(deposit=deposit) + assert deposit_request.deposit == deposit + check_archive('filename0', deposit_request.archive.name) + + # updating/adding is forbidden + + # uri to update the content + edit_se_iri = reverse( + 'edit_se_iri', args=[deposit_collection.name, deposit_id]) + em_iri = reverse( + 'em_iri', args=[deposit_collection.name, deposit_id]) + + # Testing all update/add endpoint should fail + # since the status is ready + + archive2 = create_arborescence_archive( + tmp_path, 'archive2', 'file2', b'some content in file 2') + + # replacing file is no longer possible since the deposit's + # status is ready + r = authenticated_client.put( + em_iri, + content_type='application/zip', + data=archive2['data'], + CONTENT_LENGTH=archive2['length'], + HTTP_SLUG=external_id, + HTTP_CONTENT_MD5=archive2['md5sum'], + HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', + HTTP_IN_PROGRESS='false', + HTTP_CONTENT_DISPOSITION='attachment; filename=filename0') + + assert r.status_code == status.HTTP_400_BAD_REQUEST + + # adding file is no longer possible since the deposit's status + # is ready + r = authenticated_client.post( + em_iri, + content_type='application/zip', + data=archive2['data'], + CONTENT_LENGTH=archive2['length'], + HTTP_SLUG=external_id, + HTTP_CONTENT_MD5=archive2['md5sum'], + HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', + HTTP_IN_PROGRESS='false', + HTTP_CONTENT_DISPOSITION='attachment; filename=filename0') + + assert r.status_code == status.HTTP_400_BAD_REQUEST + + # replacing metadata is no longer possible since the deposit's + # status is ready + r = authenticated_client.put( + edit_se_iri, + content_type='application/atom+xml;type=entry', + data=atom_dataset['entry-data-deposit-binary'], + CONTENT_LENGTH=len(atom_dataset['entry-data-deposit-binary']), + HTTP_SLUG=external_id) + + assert r.status_code == status.HTTP_400_BAD_REQUEST + + # adding new metadata is no longer possible since the + # deposit's status is ready + r = authenticated_client.post( + edit_se_iri, + content_type='application/atom+xml;type=entry', + data=atom_dataset['entry-data-deposit-binary'], + CONTENT_LENGTH=len(atom_dataset['entry-data-deposit-binary']), + HTTP_SLUG=external_id) + + assert r.status_code == status.HTTP_400_BAD_REQUEST + + 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) + + atom_entry = InMemoryUploadedFile( + BytesIO(atom_dataset['entry-data-deposit-binary']), + field_name='atom0', + name='atom0', + content_type='application/atom+xml; charset="utf-8"', + size=len(atom_dataset['entry-data-deposit-binary']), + charset='utf-8') + + # replacing multipart metadata is no longer possible since the + # deposit's status is ready + r = authenticated_client.put( + edit_se_iri, + format='multipart', + data={ + 'archive': archive, + 'atom_entry': atom_entry, + }) + + assert r.status_code == status.HTTP_400_BAD_REQUEST + + # adding new metadata is no longer possible since the + # deposit's status is ready + r = authenticated_client.post( + edit_se_iri, + format='multipart', + data={ + 'archive': archive, + 'atom_entry': atom_entry, + }) + + assert r.status_code == status.HTTP_400_BAD_REQUEST