Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/tests/api/test_deposit_binary.py
| # Copyright (C) 2017-2018 The Software Heritage developers | # Copyright (C) 2017-2018 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 django.core.files.uploadedfile import InMemoryUploadedFile | from django.core.files.uploadedfile import InMemoryUploadedFile | ||||
| from django.core.urlresolvers import reverse | from django.core.urlresolvers import reverse | ||||
| from io import BytesIO | from io import BytesIO | ||||
| from nose.tools import istest | |||||
| from rest_framework import status | from rest_framework import status | ||||
| from rest_framework.test import APITestCase | from rest_framework.test import APITestCase | ||||
| from swh.deposit.tests import TEST_CONFIG | from swh.deposit.tests import TEST_CONFIG | ||||
| from swh.deposit.config import COL_IRI, EM_IRI | from swh.deposit.config import COL_IRI, EM_IRI | ||||
| from swh.deposit.config import DEPOSIT_STATUS_DEPOSITED | from swh.deposit.config import DEPOSIT_STATUS_DEPOSITED | ||||
| from swh.deposit.models import Deposit, DepositRequest | from swh.deposit.models import Deposit, DepositRequest | ||||
| ▲ Show 20 Lines • Show All 102 Lines • ▼ Show 20 Lines | <entry xmlns="http://www.w3.org/2005/Atom" | ||||
| <dcterms:references>References</dcterms:references> | <dcterms:references>References</dcterms:references> | ||||
| <dcterms:rightsHolder>Rights Holder</dcterms:rightsHolder> | <dcterms:rightsHolder>Rights Holder</dcterms:rightsHolder> | ||||
| <dcterms:source>Source</dcterms:source> | <dcterms:source>Source</dcterms:source> | ||||
| <dcterms:title>Title</dcterms:title> | <dcterms:title>Title</dcterms:title> | ||||
| <dcterms:type>Type</dcterms:type> | <dcterms:type>Type</dcterms:type> | ||||
| </entry>""" | </entry>""" | ||||
| @istest | def test_post_deposit_binary_without_slug_header_is_bad_request(self): | ||||
| def post_deposit_binary_without_slug_header_is_bad_request(self): | |||||
| """Posting a binary deposit without slug header should return 400 | """Posting a binary deposit without slug header should return 400 | ||||
| """ | """ | ||||
| url = reverse(COL_IRI, args=[self.collection.name]) | url = reverse(COL_IRI, args=[self.collection.name]) | ||||
| # when | # when | ||||
| response = self.client.post( | response = self.client.post( | ||||
| url, | url, | ||||
| content_type='application/zip', # as zip | content_type='application/zip', # as zip | ||||
| data=self.archive['data'], | data=self.archive['data'], | ||||
| # + headers | # + headers | ||||
| CONTENT_LENGTH=self.archive['length'], | CONTENT_LENGTH=self.archive['length'], | ||||
| HTTP_CONTENT_MD5=self.archive['md5sum'], | HTTP_CONTENT_MD5=self.archive['md5sum'], | ||||
| HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', | HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', | ||||
| HTTP_IN_PROGRESS='false', | HTTP_IN_PROGRESS='false', | ||||
| HTTP_CONTENT_DISPOSITION='attachment; filename=filename0') | HTTP_CONTENT_DISPOSITION='attachment; filename=filename0') | ||||
| self.assertIn(b'Missing SLUG header', response.content) | self.assertIn(b'Missing SLUG header', response.content) | ||||
| self.assertEqual(response.status_code, | self.assertEqual(response.status_code, | ||||
| status.HTTP_400_BAD_REQUEST) | status.HTTP_400_BAD_REQUEST) | ||||
| @istest | def test_post_deposit_binary_upload_final_and_status_check(self): | ||||
| def post_deposit_binary_upload_final_and_status_check(self): | |||||
| """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=[self.collection.name]) | url = reverse(COL_IRI, args=[self.collection.name]) | ||||
| external_id = 'some-external-id-1' | external_id = 'some-external-id-1' | ||||
| Show All 39 Lines | def test_post_deposit_binary_upload_final_and_status_check(self): | ||||
| deposit.status) | deposit.status) | ||||
| edit_se_iri = reverse('edit_se_iri', | edit_se_iri = reverse('edit_se_iri', | ||||
| args=[self.collection.name, deposit.id]) | args=[self.collection.name, deposit.id]) | ||||
| self.assertEqual(response._headers['location'], | self.assertEqual(response._headers['location'], | ||||
| ('Location', 'http://testserver' + edit_se_iri)) | ('Location', 'http://testserver' + edit_se_iri)) | ||||
| @istest | def test_post_deposit_binary_upload_supports_zip_or_tar(self): | ||||
| def post_deposit_binary_upload_supports_zip_or_tar(self): | |||||
| """Binary upload with content-type not in [zip,x-tar] should return 415 | """Binary upload with content-type not in [zip,x-tar] should return 415 | ||||
| """ | """ | ||||
| # given | # given | ||||
| url = reverse(COL_IRI, args=[self.collection.name]) | url = reverse(COL_IRI, args=[self.collection.name]) | ||||
| external_id = 'some-external-id-1' | external_id = 'some-external-id-1' | ||||
| Show All 12 Lines | def test_post_deposit_binary_upload_supports_zip_or_tar(self): | ||||
| # then | # then | ||||
| self.assertEqual(response.status_code, | self.assertEqual(response.status_code, | ||||
| status.HTTP_415_UNSUPPORTED_MEDIA_TYPE) | status.HTTP_415_UNSUPPORTED_MEDIA_TYPE) | ||||
| with self.assertRaises(Deposit.DoesNotExist): | with self.assertRaises(Deposit.DoesNotExist): | ||||
| Deposit.objects.get(external_id=external_id) | Deposit.objects.get(external_id=external_id) | ||||
| @istest | def test_post_deposit_binary_fails_if_unsupported_packaging_header( | ||||
| def post_deposit_binary_fails_if_unsupported_packaging_header( | |||||
| self): | self): | ||||
| """Bin deposit without supported content_disposition header returns 400 | """Bin deposit without supported content_disposition header returns 400 | ||||
| """ | """ | ||||
| # given | # given | ||||
| url = reverse(COL_IRI, args=[self.collection.name]) | url = reverse(COL_IRI, args=[self.collection.name]) | ||||
| external_id = 'some-external-id' | external_id = 'some-external-id' | ||||
| Show All 11 Lines | def test_post_deposit_binary_fails_if_unsupported_packaging_header( | ||||
| HTTP_CONTENT_DISPOSITION='attachment; filename=filename0') | HTTP_CONTENT_DISPOSITION='attachment; filename=filename0') | ||||
| # then | # then | ||||
| self.assertEqual(response.status_code, | self.assertEqual(response.status_code, | ||||
| status.HTTP_400_BAD_REQUEST) | status.HTTP_400_BAD_REQUEST) | ||||
| with self.assertRaises(Deposit.DoesNotExist): | with self.assertRaises(Deposit.DoesNotExist): | ||||
| Deposit.objects.get(external_id=external_id) | Deposit.objects.get(external_id=external_id) | ||||
| @istest | def test_post_deposit_binary_upload_fail_if_no_content_disposition_header( | ||||
| def post_deposit_binary_upload_fail_if_no_content_disposition_header( | |||||
| self): | self): | ||||
| """Binary upload without content_disposition header should return 400 | """Binary upload without content_disposition header should return 400 | ||||
| """ | """ | ||||
| # given | # given | ||||
| url = reverse(COL_IRI, args=[self.collection.name]) | url = reverse(COL_IRI, args=[self.collection.name]) | ||||
| external_id = 'some-external-id' | external_id = 'some-external-id' | ||||
| Show All 11 Lines | def test_post_deposit_binary_upload_fail_if_no_content_disposition_header( | ||||
| HTTP_IN_PROGRESS='false') | HTTP_IN_PROGRESS='false') | ||||
| # then | # then | ||||
| self.assertEqual(response.status_code, | self.assertEqual(response.status_code, | ||||
| status.HTTP_400_BAD_REQUEST) | status.HTTP_400_BAD_REQUEST) | ||||
| with self.assertRaises(Deposit.DoesNotExist): | with self.assertRaises(Deposit.DoesNotExist): | ||||
| Deposit.objects.get(external_id=external_id) | Deposit.objects.get(external_id=external_id) | ||||
| @istest | def test_post_deposit_mediation_not_supported(self): | ||||
| def post_deposit_mediation_not_supported(self): | |||||
| """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=[self.collection.name]) | url = reverse(COL_IRI, args=[self.collection.name]) | ||||
| external_id = 'some-external-id-1' | external_id = 'some-external-id-1' | ||||
| Show All 13 Lines | def test_post_deposit_mediation_not_supported(self): | ||||
| # then | # then | ||||
| self.assertEqual(response.status_code, | self.assertEqual(response.status_code, | ||||
| status.HTTP_412_PRECONDITION_FAILED) | status.HTTP_412_PRECONDITION_FAILED) | ||||
| with self.assertRaises(Deposit.DoesNotExist): | with self.assertRaises(Deposit.DoesNotExist): | ||||
| Deposit.objects.get(external_id=external_id) | Deposit.objects.get(external_id=external_id) | ||||
| @istest | def test_post_deposit_binary_upload_fail_if_upload_size_limit_exceeded( | ||||
| def post_deposit_binary_upload_fail_if_upload_size_limit_exceeded( | |||||
| self): | self): | ||||
| """Binary upload must not exceed the limit set up... | """Binary upload must not exceed the limit set up... | ||||
| """ | """ | ||||
| # given | # given | ||||
| url = reverse(COL_IRI, args=[self.collection.name]) | url = reverse(COL_IRI, args=[self.collection.name]) | ||||
| archive = create_arborescence_archive( | archive = create_arborescence_archive( | ||||
| Show All 18 Lines | def test_post_deposit_binary_upload_fail_if_upload_size_limit_exceeded( | ||||
| # then | # then | ||||
| self.assertEqual(response.status_code, | self.assertEqual(response.status_code, | ||||
| status.HTTP_413_REQUEST_ENTITY_TOO_LARGE) | status.HTTP_413_REQUEST_ENTITY_TOO_LARGE) | ||||
| self.assertRegex(response.content, b'Upload size limit exceeded') | self.assertRegex(response.content, b'Upload size limit exceeded') | ||||
| with self.assertRaises(Deposit.DoesNotExist): | with self.assertRaises(Deposit.DoesNotExist): | ||||
| Deposit.objects.get(external_id=external_id) | Deposit.objects.get(external_id=external_id) | ||||
| @istest | def test_post_deposit_2_post_2_different_deposits(self): | ||||
| def post_deposit_2_post_2_different_deposits(self): | |||||
| """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=[self.collection.name]) | url = reverse(COL_IRI, args=[self.collection.name]) | ||||
| # when | # when | ||||
| response = self.client.post( | response = self.client.post( | ||||
| url, | url, | ||||
| Show All 40 Lines | def test_post_deposit_2_post_2_different_deposits(self): | ||||
| deposit2 = Deposit.objects.get(pk=deposit_id2) | deposit2 = Deposit.objects.get(pk=deposit_id2) | ||||
| self.assertNotEqual(deposit, deposit2) | self.assertNotEqual(deposit, deposit2) | ||||
| deposits = Deposit.objects.all().order_by('id') | deposits = Deposit.objects.all().order_by('id') | ||||
| self.assertEqual(len(deposits), 2) | self.assertEqual(len(deposits), 2) | ||||
| self.assertEqual(list(deposits), [deposit, deposit2]) | self.assertEqual(list(deposits), [deposit, deposit2]) | ||||
| @istest | def test_post_deposit_binary_and_post_to_add_another_archive(self): | ||||
| def post_deposit_binary_and_post_to_add_another_archive(self): | |||||
| """Updating a deposit should return a 201 with receipt | """Updating a deposit should return a 201 with receipt | ||||
| """ | """ | ||||
| # given | # given | ||||
| url = reverse(COL_IRI, args=[self.collection.name]) | url = reverse(COL_IRI, args=[self.collection.name]) | ||||
| external_id = 'some-external-id-1' | external_id = 'some-external-id-1' | ||||
| ▲ Show 20 Lines • Show All 73 Lines • ▼ Show 20 Lines | def test_post_deposit_binary_and_post_to_add_another_archive(self): | ||||
| self.assertEquals(deposit_requests[1].type.name, 'archive') | self.assertEquals(deposit_requests[1].type.name, 'archive') | ||||
| self.assertRegex(deposit_requests[1].archive.name, | self.assertRegex(deposit_requests[1].archive.name, | ||||
| archive2['name']) | archive2['name']) | ||||
| # only 1 deposit in db | # only 1 deposit in db | ||||
| deposits = Deposit.objects.all() | deposits = Deposit.objects.all() | ||||
| self.assertEqual(len(deposits), 1) | self.assertEqual(len(deposits), 1) | ||||
| @istest | def test_post_deposit_then_post_or_put_is_refused_when_status_ready(self): | ||||
| def post_deposit_then_post_or_put_is_refused_when_status_ready(self): | |||||
| """Updating a deposit with status 'ready' should return a 400 | """Updating a deposit with status 'ready' should return a 400 | ||||
| """ | """ | ||||
| url = reverse(COL_IRI, args=[self.collection.name]) | url = reverse(COL_IRI, args=[self.collection.name]) | ||||
| external_id = 'some-external-id-1' | external_id = 'some-external-id-1' | ||||
| # when | # when | ||||
| ▲ Show 20 Lines • Show All 135 Lines • Show Last 20 Lines | |||||