Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/tests/api/test_deposit.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 | ||||
import hashlib | import hashlib | ||||
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, nottest | |||||
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.config import COL_IRI, EDIT_SE_IRI, DEPOSIT_STATUS_REJECTED | from swh.deposit.config import COL_IRI, EDIT_SE_IRI, DEPOSIT_STATUS_REJECTED | ||||
from swh.deposit.config import DEPOSIT_STATUS_PARTIAL | from swh.deposit.config import DEPOSIT_STATUS_PARTIAL | ||||
from swh.deposit.config import DEPOSIT_STATUS_LOAD_SUCCESS | from swh.deposit.config import DEPOSIT_STATUS_LOAD_SUCCESS | ||||
from swh.deposit.config import DEPOSIT_STATUS_LOAD_FAILURE | from swh.deposit.config import DEPOSIT_STATUS_LOAD_FAILURE | ||||
from swh.deposit.models import Deposit, DepositClient, DepositCollection | from swh.deposit.models import Deposit, DepositClient, DepositCollection | ||||
from swh.deposit.parsers import parse_xml | from swh.deposit.parsers import parse_xml | ||||
from ..common import BasicTestCase, WithAuthTestCase, CommonCreationRoutine | from ..common import BasicTestCase, WithAuthTestCase, CommonCreationRoutine | ||||
class DepositNoAuthCase(APITestCase, BasicTestCase): | class DepositNoAuthCase(APITestCase, BasicTestCase): | ||||
"""Deposit access are protected with basic authentication. | """Deposit access are protected with basic authentication. | ||||
""" | """ | ||||
@istest | def test_post_will_fail_with_401(self): | ||||
def post_will_fail_with_401(self): | |||||
"""Without authentication, endpoint refuses access with 401 response | """Without authentication, endpoint refuses access with 401 response | ||||
""" | """ | ||||
url = reverse(COL_IRI, args=[self.collection.name]) | url = reverse(COL_IRI, args=[self.collection.name]) | ||||
# when | # when | ||||
response = self.client.post(url) | response = self.client.post(url) | ||||
Show All 11 Lines | def setUp(self): | ||||
# Add another user | # Add another user | ||||
_collection2 = DepositCollection(name='some') | _collection2 = DepositCollection(name='some') | ||||
_collection2.save() | _collection2.save() | ||||
_user = DepositClient.objects.create_user(username='user', | _user = DepositClient.objects.create_user(username='user', | ||||
password='user') | password='user') | ||||
_user.collections = [_collection2.id] | _user.collections = [_collection2.id] | ||||
self.collection2 = _collection2 | self.collection2 = _collection2 | ||||
@istest | def test_access_to_another_user_collection_is_forbidden(self): | ||||
def access_to_another_user_collection_is_forbidden(self): | |||||
"""Access to another user collection should return a 403 | """Access to another user collection should return a 403 | ||||
""" | """ | ||||
url = reverse(COL_IRI, args=[self.collection2.name]) | url = reverse(COL_IRI, args=[self.collection2.name]) | ||||
response = self.client.post(url) | response = self.client.post(url) | ||||
self.assertEqual(response.status_code, | self.assertEqual(response.status_code, | ||||
status.HTTP_403_FORBIDDEN) | status.HTTP_403_FORBIDDEN) | ||||
self.assertRegex(response.content.decode('utf-8'), | self.assertRegex(response.content.decode('utf-8'), | ||||
'Client hal cannot access collection %s' % ( | 'Client hal cannot access collection %s' % ( | ||||
self.collection2.name, )) | self.collection2.name, )) | ||||
@istest | def test_delete_on_col_iri_not_supported(self): | ||||
def delete_on_col_iri_not_supported(self): | |||||
"""Delete on col iri should return a 405 response | """Delete on col iri should return a 405 response | ||||
""" | """ | ||||
url = reverse(COL_IRI, args=[self.collection.name]) | url = reverse(COL_IRI, args=[self.collection.name]) | ||||
response = self.client.delete(url) | response = self.client.delete(url) | ||||
self.assertEqual(response.status_code, | self.assertEqual(response.status_code, | ||||
status.HTTP_405_METHOD_NOT_ALLOWED) | status.HTTP_405_METHOD_NOT_ALLOWED) | ||||
self.assertRegex(response.content.decode('utf-8'), | self.assertRegex(response.content.decode('utf-8'), | ||||
'DELETE method is not supported on this endpoint') | 'DELETE method is not supported on this endpoint') | ||||
@nottest | |||||
def create_deposit_with_rejection_status(self): | def create_deposit_with_rejection_status(self): | ||||
url = reverse(COL_IRI, args=[self.collection.name]) | url = reverse(COL_IRI, args=[self.collection.name]) | ||||
data = b'some data which is clearly not a zip file' | data = b'some data which is clearly not a zip file' | ||||
md5sum = hashlib.md5(data).hexdigest() | md5sum = hashlib.md5(data).hexdigest() | ||||
external_id = 'some-external-id-1' | external_id = 'some-external-id-1' | ||||
# when | # when | ||||
Show All 9 Lines | def create_deposit_with_rejection_status(self): | ||||
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') | ||||
self.assertEquals(response.status_code, status.HTTP_201_CREATED) | self.assertEquals(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'] | ||||
self.assertEquals(actual_state, DEPOSIT_STATUS_REJECTED) | self.assertEquals(actual_state, DEPOSIT_STATUS_REJECTED) | ||||
@istest | def test_act_on_deposit_rejected_is_not_permitted(self): | ||||
def act_on_deposit_rejected_is_not_permitted(self): | |||||
deposit_id = self.create_deposit_with_status(DEPOSIT_STATUS_REJECTED) | deposit_id = self.create_deposit_with_status(DEPOSIT_STATUS_REJECTED) | ||||
deposit = Deposit.objects.get(pk=deposit_id) | deposit = Deposit.objects.get(pk=deposit_id) | ||||
assert deposit.status == DEPOSIT_STATUS_REJECTED | assert deposit.status == DEPOSIT_STATUS_REJECTED | ||||
response = self.client.post( | response = self.client.post( | ||||
reverse(EDIT_SE_IRI, args=[self.collection.name, deposit_id]), | reverse(EDIT_SE_IRI, args=[self.collection.name, deposit_id]), | ||||
content_type='application/atom+xml;type=entry', | content_type='application/atom+xml;type=entry', | ||||
data=self.atom_entry_data1, | data=self.atom_entry_data1, | ||||
HTTP_SLUG='external-id') | HTTP_SLUG='external-id') | ||||
self.assertEquals(response.status_code, status.HTTP_400_BAD_REQUEST) | self.assertEquals(response.status_code, status.HTTP_400_BAD_REQUEST) | ||||
self.assertRegex( | self.assertRegex( | ||||
response.content.decode('utf-8'), | response.content.decode('utf-8'), | ||||
"You can only act on deposit with status '%s'" % ( | "You can only act on deposit with status '%s'" % ( | ||||
DEPOSIT_STATUS_PARTIAL, )) | DEPOSIT_STATUS_PARTIAL, )) | ||||
@istest | def test_add_deposit_with_parent(self): | ||||
def add_deposit_with_parent(self): | |||||
# given multiple deposit already loaded | # given multiple deposit already loaded | ||||
deposit_id = self.create_deposit_with_status( | deposit_id = self.create_deposit_with_status( | ||||
status=DEPOSIT_STATUS_LOAD_SUCCESS, | status=DEPOSIT_STATUS_LOAD_SUCCESS, | ||||
external_id='some-external-id') | external_id='some-external-id') | ||||
deposit1 = Deposit.objects.get(pk=deposit_id) | deposit1 = Deposit.objects.get(pk=deposit_id) | ||||
self.assertIsNotNone(deposit1) | self.assertIsNotNone(deposit1) | ||||
self.assertEquals(deposit1.external_id, 'some-external-id') | self.assertEquals(deposit1.external_id, 'some-external-id') | ||||
Show All 31 Lines |