Page MenuHomeSoftware Heritage
Differential D378 Diff 1161 swh/deposit/tests/common.py

Changeset View

Standalone View

View Options

swh/deposit/tests/common.py

Show All 9 Lines
import tempfile import tempfile
from django.core.urlresolvers import reverse from django.core.urlresolvers import reverse
from django.test import TestCase from django.test import TestCase
from io import BytesIO from io import BytesIO
from nose.plugins.attrib import attr from nose.plugins.attrib import attr
from rest_framework import status from rest_framework import status
from swh.deposit.config import COL_IRI, EM_IRI, EDIT_SE_IRI from swh.deposit.config import (COL_IRI, EM_IRI, EDIT_SE_IRI,
DEPOSIT_STATUS_PARTIAL,
DEPOSIT_STATUS_VERIFIED,
DEPOSIT_STATUS_DEPOSITED)
from swh.deposit.models import DepositClient, DepositCollection, Deposit from swh.deposit.models import DepositClient, DepositCollection, Deposit
from swh.deposit.models import DepositRequest from swh.deposit.models import DepositRequest
from swh.deposit.models import DepositRequestType from swh.deposit.models import DepositRequestType
from swh.deposit.parsers import parse_xml from swh.deposit.parsers import parse_xml
from swh.deposit.settings.testing import MEDIA_ROOT from swh.deposit.settings.testing import MEDIA_ROOT
from swh.core import tarball from swh.core import tarball
▲ Show 20 Lines • Show All 89 Lines • ▼ Show 20 Lines def create_simple_binary_deposit(self, status_partial=True):
CONTENT_LENGTH=self.archive['length'], CONTENT_LENGTH=self.archive['length'],
HTTP_MD5SUM=self.archive['md5sum'], HTTP_MD5SUM=self.archive['md5sum'],
HTTP_SLUG='external-id', HTTP_SLUG='external-id',
HTTP_IN_PROGRESS=status_partial, HTTP_IN_PROGRESS=status_partial,
HTTP_CONTENT_DISPOSITION='attachment; filename=%s' % ( HTTP_CONTENT_DISPOSITION='attachment; filename=%s' % (
self.archive['name'], )) self.archive['name'], ))
# then # then
assert response.status_code == status.HTTP_201_CREATED self.assertEqual(response.status_code, status.HTTP_201_CREATED)
response_content = parse_xml(BytesIO(response.content)) response_content = parse_xml(BytesIO(response.content))
_status = response_content['deposit_status']
if status_partial:
expected_status = DEPOSIT_STATUS_PARTIAL
else:
expected_status = DEPOSIT_STATUS_VERIFIED
self.assertEqual(_status, expected_status)
deposit_id = int(response_content['deposit_id']) deposit_id = int(response_content['deposit_id'])
return deposit_id return deposit_id
def create_complex_binary_deposit(self, status_partial=False): def create_complex_binary_deposit(self, status_partial=False):
deposit_id = self.create_simple_binary_deposit( deposit_id = self.create_simple_binary_deposit(
status_partial=True) status_partial=True)
# Add a second archive to the deposit # Add a second archive to the deposit
Show All 19 Lines def update_binary_deposit(self, deposit_id, status_partial=False):
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.codemeta_entry_data1, data=self.codemeta_entry_data1,
HTTP_SLUG='external-id', HTTP_SLUG='external-id',
HTTP_IN_PROGRESS=status_partial) HTTP_IN_PROGRESS=status_partial)
# then # then
# assert response.status_code == status.HTTP_201_CREATED self.assertEqual(response.status_code, status.HTTP_201_CREATED)
response_content = parse_xml(BytesIO(response.content)) response_content = parse_xml(BytesIO(response.content))
_status = response_content['deposit_status']
if status_partial:
expected_status = DEPOSIT_STATUS_PARTIAL
else:
expected_status = DEPOSIT_STATUS_DEPOSITED
self.assertEqual(_status, expected_status)
deposit_id = int(response_content['deposit_id']) deposit_id = int(response_content['deposit_id'])
return deposit_id return deposit_id
@attr('fs') @attr('fs')
class BasicTestCase(TestCase): class BasicTestCase(TestCase):
"""Mixin intended for data setup purposes (user, collection, etc...) """Mixin intended for data setup purposes (user, collection, etc...)
▲ Show 20 Lines • Show All 137 Lines • ▼ Show 20 Lines <author>
<name>HAL</name> <name>HAL</name>
<email>hal@ccsd.cnrs.fr</email> <email>hal@ccsd.cnrs.fr</email>
</author> </author>
<codemeta:author> <codemeta:author>
<codemeta:name>Morane Gruenpeter</codemeta:name> <codemeta:name>Morane Gruenpeter</codemeta:name>
</codemeta:author> </codemeta:author>
</entry>""" </entry>"""
def create_invalid_deposit(self, external_id='some-external-id-1'): def create_deposit_with_invalid_archive(self,
external_id='some-external-id-1'):
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()
# 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=data, data=data,
# + headers # + headers
CONTENT_LENGTH=len(data), CONTENT_LENGTH=len(data),
# other headers needs HTTP_ prefix to be taken into account # other headers needs HTTP_ prefix to be taken into account
HTTP_SLUG=external_id, HTTP_SLUG=external_id,
HTTP_CONTENT_MD5=md5sum, HTTP_CONTENT_MD5=md5sum,
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')
response_content = parse_xml(BytesIO(response.content)) response_content = parse_xml(BytesIO(response.content))
deposit_id = int(response_content['deposit_id']) deposit_id = int(response_content['deposit_id'])
return deposit_id return deposit_id
def create_deposit_with_status( def create_deposit_with_status(
self, status, external_id='some-external-id-1', swh_id=None): self, status, external_id='some-external-id-1', swh_id=None):
deposit_id = self.create_invalid_deposit(external_id) # create an invalid deposit which we will update further down the line
deposit_id = self.create_deposit_with_invalid_archive(external_id)
# We cannot create some form of deposit with a given status in # We cannot create some form of deposit with a given status in
# test context ('rejected' for example). As flipped off the # test context ('rejected' for example). Update in place the
# checks in the configuration so all deposits have the status # deposit with such status to permit some further tests.
# deposited). Update in place the deposit with such
# status
deposit = Deposit.objects.get(pk=deposit_id) deposit = Deposit.objects.get(pk=deposit_id)
deposit.status = status deposit.status = status
if swh_id: if swh_id:
deposit.swh_id = swh_id deposit.swh_id = swh_id
deposit.save() deposit.save()
return deposit_id return deposit_id
▲ Show 20 Lines • Show All 105 Lines • Show Last 20 Lines
About · Developer information · Legal