Page MenuHomeSoftware Heritage
Files F9123448

No OneTemporary
Actions

View Options

diff --git a/swh/deposit/api/common.py b/swh/deposit/api/common.py
index 6573844f..b83846d8 100644
--- a/swh/deposit/api/common.py
+++ b/swh/deposit/api/common.py
@@ -1,877 +1,877 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# 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 hashlib
from abc import ABCMeta, abstractmethod
-from django.core.urlresolvers import reverse
+from django.urls import reverse
from django.http import HttpResponse
from django.shortcuts import render
from django.utils import timezone
from rest_framework import status
from rest_framework.authentication import BasicAuthentication
from rest_framework.permissions import IsAuthenticated, AllowAny
from rest_framework.views import APIView
from swh.model import hashutil
from ..config import (
SWHDefaultConfig, EDIT_SE_IRI, EM_IRI, CONT_FILE_IRI,
ARCHIVE_KEY, METADATA_KEY, RAW_METADATA_KEY, STATE_IRI,
DEPOSIT_STATUS_DEPOSITED, DEPOSIT_STATUS_PARTIAL,
DEPOSIT_STATUS_LOAD_SUCCESS, ARCHIVE_TYPE, METADATA_TYPE
)
from ..errors import (
MAX_UPLOAD_SIZE_EXCEEDED, BAD_REQUEST, ERROR_CONTENT,
CHECKSUM_MISMATCH, make_error_dict, MEDIATION_NOT_ALLOWED,
make_error_response_from_dict, FORBIDDEN,
NOT_FOUND, make_error_response, METHOD_NOT_ALLOWED
)
from ..models import (
Deposit, DepositRequest, DepositCollection,
DepositClient
)
from ..parsers import parse_xml
ACCEPT_PACKAGINGS = ['http://purl.org/net/sword/package/SimpleZip']
ACCEPT_ARCHIVE_CONTENT_TYPES = ['application/zip', 'application/x-tar']
class SWHAPIView(APIView):
"""Mixin intended as a based API view to enforce the basic
authentication check
"""
authentication_classes = (BasicAuthentication, )
permission_classes = (IsAuthenticated, )
class SWHPrivateAPIView(SWHAPIView):
"""Mixin intended as private api (so no authentication) based API view
(for the private ones).
"""
authentication_classes = ()
permission_classes = (AllowAny, )
class SWHBaseDeposit(SWHDefaultConfig, SWHAPIView, metaclass=ABCMeta):
"""Base deposit request class sharing multiple common behaviors.
"""
def _read_headers(self, req):
"""Read and unify the necessary headers from the request (those are
not stored in the same location or not properly formatted).
Args:
req (Request): Input request
Returns:
Dictionary with the following keys (some associated values may be
None):
- content-type
- content-length
- in-progress
- content-disposition
- packaging
- slug
- on-behalf-of
"""
meta = req._request.META
content_type = req.content_type
content_length = meta.get('CONTENT_LENGTH')
if content_length and isinstance(content_length, str):
content_length = int(content_length)
# final deposit if not provided
in_progress = meta.get('HTTP_IN_PROGRESS', False)
content_disposition = meta.get('HTTP_CONTENT_DISPOSITION')
if isinstance(in_progress, str):
in_progress = in_progress.lower() == 'true'
content_md5sum = meta.get('HTTP_CONTENT_MD5')
if content_md5sum:
content_md5sum = bytes.fromhex(content_md5sum)
packaging = meta.get('HTTP_PACKAGING')
slug = meta.get('HTTP_SLUG')
on_behalf_of = meta.get('HTTP_ON_BEHALF_OF')
metadata_relevant = meta.get('HTTP_METADATA_RELEVANT')
return {
'content-type': content_type,
'content-length': content_length,
'in-progress': in_progress,
'content-disposition': content_disposition,
'content-md5sum': content_md5sum,
'packaging': packaging,
'slug': slug,
'on-behalf-of': on_behalf_of,
'metadata-relevant': metadata_relevant,
}
def _compute_md5(self, filehandler):
"""Compute uploaded file's md5 sum.
Args:
filehandler (InMemoryUploadedFile): the file to compute the md5
hash
Returns:
the md5 checksum (str)
"""
h = hashlib.md5()
for chunk in filehandler:
h.update(chunk)
return h.digest()
def _deposit_put(self, deposit_id=None, in_progress=False,
external_id=None):
"""Save/Update a deposit in db.
Args:
deposit_id (int): deposit identifier
in_progress (dict): The deposit's status
external_id (str): The external identifier to associate to
the deposit
Returns:
The Deposit instance saved or updated.
"""
if in_progress is False:
complete_date = timezone.now()
status_type = DEPOSIT_STATUS_DEPOSITED
else:
complete_date = None
status_type = DEPOSIT_STATUS_PARTIAL
if not deposit_id:
try:
# find a deposit parent (same external id, status load
# to success)
deposit_parent = Deposit.objects.filter(
external_id=external_id,
status=DEPOSIT_STATUS_LOAD_SUCCESS).order_by('-id')[0:1].get() # noqa
except Deposit.DoesNotExist:
deposit_parent = None
deposit = Deposit(collection=self._collection,
external_id=external_id,
complete_date=complete_date,
status=status_type,
client=self._client,
parent=deposit_parent)
else:
deposit = Deposit.objects.get(pk=deposit_id)
# update metadata
deposit.complete_date = complete_date
deposit.status = status_type
deposit.save()
return deposit
def _deposit_request_put(self, deposit, deposit_request_data,
replace_metadata=False, replace_archives=False):
"""Save a deposit request with metadata attached to a deposit.
Args:
deposit (Deposit): The deposit concerned by the request
deposit_request_data (dict): The dictionary with at most 2 deposit
request types (archive, metadata) to associate to the deposit
replace_metadata (bool): Flag defining if we add or update
existing metadata to the deposit
replace_archives (bool): Flag defining if we add or update
archives to existing deposit
Returns:
None
"""
if replace_metadata:
DepositRequest.objects.filter(
deposit=deposit,
type=METADATA_TYPE).delete()
if replace_archives:
DepositRequest.objects.filter(
deposit=deposit,
type=ARCHIVE_TYPE).delete()
deposit_request = None
archive_file = deposit_request_data.get(ARCHIVE_KEY)
if archive_file:
deposit_request = DepositRequest(
type=ARCHIVE_TYPE,
deposit=deposit,
archive=archive_file)
deposit_request.save()
metadata = deposit_request_data.get(METADATA_KEY)
if metadata:
raw_metadata = deposit_request_data.get(RAW_METADATA_KEY)
deposit_request = DepositRequest(
type=METADATA_TYPE,
deposit=deposit,
metadata=metadata,
raw_metadata=raw_metadata)
deposit_request.save()
assert deposit_request is not None
def _delete_archives(self, collection_name, deposit_id):
"""Delete archives reference from the deposit id.
"""
try:
deposit = Deposit.objects.get(pk=deposit_id)
except Deposit.DoesNotExist:
return make_error_dict(
NOT_FOUND,
'The deposit %s does not exist' % deposit_id)
DepositRequest.objects.filter(
deposit=deposit,
type=ARCHIVE_TYPE).delete()
return {}
def _delete_deposit(self, collection_name, deposit_id):
"""Delete deposit reference.
Args:
collection_name (str): Client's name
deposit_id (id): The deposit to delete
Returns
Empty dict when ok.
Dict with error key to describe the failure.
"""
try:
deposit = Deposit.objects.get(pk=deposit_id)
except Deposit.DoesNotExist:
return make_error_dict(
NOT_FOUND,
'The deposit %s does not exist' % deposit_id)
if deposit.collection.name != collection_name:
summary = 'Cannot delete a deposit from another collection'
description = "Deposit %s does not belong to the collection %s" % (
deposit_id, collection_name)
return make_error_dict(
BAD_REQUEST,
summary=summary,
verbose_description=description)
DepositRequest.objects.filter(deposit=deposit).delete()
deposit.delete()
return {}
def _check_preconditions_on(self, filehandler, md5sum,
content_length=None):
"""Check preconditions on provided file are respected. That is the
length and/or the md5sum hash match the file's content.
Args:
filehandler (InMemoryUploadedFile): The file to check
md5sum (hex str): md5 hash expected from the file's content
content_length (int): the expected length if provided.
Returns:
Either none if no error or a dictionary with a key error
detailing the problem.
"""
if content_length:
if content_length > self.config['max_upload_size']:
return make_error_dict(
MAX_UPLOAD_SIZE_EXCEEDED,
'Upload size limit exceeded (max %s bytes).' %
self.config['max_upload_size'],
'Please consider sending the archive in '
'multiple steps.')
length = filehandler.size
if length != content_length:
return make_error_dict(status.HTTP_412_PRECONDITION_FAILED,
'Wrong length')
if md5sum:
_md5sum = self._compute_md5(filehandler)
if _md5sum != md5sum:
return make_error_dict(
CHECKSUM_MISMATCH,
'Wrong md5 hash',
'The checksum sent %s and the actual checksum '
'%s does not match.' % (hashutil.hash_to_hex(md5sum),
hashutil.hash_to_hex(_md5sum)))
return None
def _binary_upload(self, req, headers, collection_name, deposit_id=None,
replace_metadata=False, replace_archives=False):
"""Binary upload routine.
Other than such a request, a 415 response is returned.
Args:
req (Request): the request holding information to parse
and inject in db
headers (dict): request headers formatted
collection_name (str): the associated client
deposit_id (id): deposit identifier if provided
replace_metadata (bool): 'Update or add' request to existing
deposit. If False (default), this adds new metadata request to
existing ones. Otherwise, this will replace existing metadata.
replace_archives (bool): 'Update or add' request to existing
deposit. If False (default), this adds new archive request to
existing ones. Otherwise, this will replace existing archives.
ones.
Returns:
In the optimal case a dict with the following keys:
- deposit_id (int): Deposit identifier
- deposit_date (date): Deposit date
- archive: None (no archive is provided here)
Otherwise, a dictionary with the key error and the
associated failures, either:
- 400 (bad request) if the request is not providing an external
identifier
- 413 (request entity too large) if the length of the
archive exceeds the max size configured
- 412 (precondition failed) if the length or md5 hash provided
mismatch the reality of the archive
- 415 (unsupported media type) if a wrong media type is provided
"""
content_length = headers['content-length']
if not content_length:
return make_error_dict(
BAD_REQUEST,
'CONTENT_LENGTH header is mandatory',
'For archive deposit, the '
'CONTENT_LENGTH header must be sent.')
content_disposition = headers['content-disposition']
if not content_disposition:
return make_error_dict(
BAD_REQUEST,
'CONTENT_DISPOSITION header is mandatory',
'For archive deposit, the '
'CONTENT_DISPOSITION header must be sent.')
packaging = headers['packaging']
if packaging and packaging not in ACCEPT_PACKAGINGS:
return make_error_dict(
BAD_REQUEST,
'Only packaging %s is supported' %
ACCEPT_PACKAGINGS,
'The packaging provided %s is not supported' % packaging)
filehandler = req.FILES['file']
precondition_status_response = self._check_preconditions_on(
filehandler, headers['content-md5sum'], content_length)
if precondition_status_response:
return precondition_status_response
external_id = headers['slug']
# actual storage of data
archive_metadata = filehandler
deposit = self._deposit_put(deposit_id=deposit_id,
in_progress=headers['in-progress'],
external_id=external_id)
self._deposit_request_put(
deposit, {ARCHIVE_KEY: archive_metadata},
replace_metadata=replace_metadata,
replace_archives=replace_archives)
return {
'deposit_id': deposit.id,
'deposit_date': deposit.reception_date,
'status': deposit.status,
'archive': filehandler.name,
}
def _read_metadata(self, metadata_stream):
"""Given a metadata stream, reads the metadata and returns both the
parsed and the raw metadata.
"""
raw_metadata = metadata_stream.read()
metadata = parse_xml(raw_metadata)
return raw_metadata, metadata
def _multipart_upload(self, req, headers, collection_name,
deposit_id=None, replace_metadata=False,
replace_archives=False):
"""Multipart upload supported with exactly:
- 1 archive (zip)
- 1 atom entry
Other than such a request, a 415 response is returned.
Args:
req (Request): the request holding information to parse
and inject in db
headers (dict): request headers formatted
collection_name (str): the associated client
deposit_id (id): deposit identifier if provided
replace_metadata (bool): 'Update or add' request to existing
deposit. If False (default), this adds new metadata request to
existing ones. Otherwise, this will replace existing metadata.
replace_archives (bool): 'Update or add' request to existing
deposit. If False (default), this adds new archive request to
existing ones. Otherwise, this will replace existing archives.
ones.
Returns:
In the optimal case a dict with the following keys:
- deposit_id (int): Deposit identifier
- deposit_date (date): Deposit date
- archive: None (no archive is provided here)
Otherwise, a dictionary with the key error and the
associated failures, either:
- 400 (bad request) if the request is not providing an external
identifier
- 412 (precondition failed) if the potentially md5 hash provided
mismatch the reality of the archive
- 413 (request entity too large) if the length of the
archive exceeds the max size configured
- 415 (unsupported media type) if a wrong media type is provided
"""
external_id = headers['slug']
content_types_present = set()
data = {
'application/zip': None, # expected either zip
'application/x-tar': None, # or x-tar
'application/atom+xml': None,
}
for key, value in req.FILES.items():
fh = value
if fh.content_type in content_types_present:
return make_error_dict(
ERROR_CONTENT,
'Only 1 application/zip (or application/x-tar) archive '
'and 1 atom+xml entry is supported (as per sword2.0 '
'specification)',
'You provided more than 1 application/(zip|x-tar) '
'or more than 1 application/atom+xml content-disposition '
'header in the multipart deposit')
content_types_present.add(fh.content_type)
data[fh.content_type] = fh
if len(content_types_present) != 2:
return make_error_dict(
ERROR_CONTENT,
'You must provide both 1 application/zip (or '
'application/x-tar) and 1 atom+xml entry for multipart '
'deposit',
'You need to provide only 1 application/(zip|x-tar) '
'and 1 application/atom+xml content-disposition header '
'in the multipart deposit')
filehandler = data['application/zip']
if not filehandler:
filehandler = data['application/x-tar']
precondition_status_response = self._check_preconditions_on(
filehandler,
headers['content-md5sum'])
if precondition_status_response:
return precondition_status_response
raw_metadata, metadata = self._read_metadata(
data['application/atom+xml'])
# actual storage of data
deposit = self._deposit_put(deposit_id=deposit_id,
in_progress=headers['in-progress'],
external_id=external_id)
deposit_request_data = {
ARCHIVE_KEY: filehandler,
METADATA_KEY: metadata,
RAW_METADATA_KEY: raw_metadata,
}
self._deposit_request_put(
deposit, deposit_request_data, replace_metadata, replace_archives)
return {
'deposit_id': deposit.id,
'deposit_date': deposit.reception_date,
'archive': filehandler.name,
'status': deposit.status,
}
def _atom_entry(self, req, headers, collection_name,
deposit_id=None,
replace_metadata=False,
replace_archives=False):
"""Atom entry deposit.
Args:
req (Request): the request holding information to parse
and inject in db
headers (dict): request headers formatted
collection_name (str): the associated client
deposit_id (id): deposit identifier if provided
replace_metadata (bool): 'Update or add' request to existing
deposit. If False (default), this adds new metadata request to
existing ones. Otherwise, this will replace existing metadata.
replace_archives (bool): 'Update or add' request to existing
deposit. If False (default), this adds new archive request to
existing ones. Otherwise, this will replace existing archives.
ones.
Returns:
In the optimal case a dict with the following keys:
- deposit_id: deposit id associated to the deposit
- deposit_date: date of the deposit
- archive: None (no archive is provided here)
Otherwise, a dictionary with the key error and the
associated failures, either:
- 400 (bad request) if the request is not providing an external
identifier
- 400 (bad request) if the request's body is empty
- 415 (unsupported media type) if a wrong media type is provided
"""
raw_metadata, metadata = self._read_metadata(req.data)
if not metadata:
return make_error_dict(
BAD_REQUEST,
'Empty body request is not supported',
'Atom entry deposit is supposed to send for metadata. '
'If the body is empty, there is no metadata.')
external_id = metadata.get('external_identifier', headers['slug'])
deposit = self._deposit_put(deposit_id=deposit_id,
in_progress=headers['in-progress'],
external_id=external_id)
self._deposit_request_put(
deposit, {METADATA_KEY: metadata, RAW_METADATA_KEY: raw_metadata},
replace_metadata, replace_archives)
return {
'deposit_id': deposit.id,
'deposit_date': deposit.reception_date,
'archive': None,
'status': deposit.status,
}
def _empty_post(self, req, headers, collection_name, deposit_id):
"""Empty post to finalize an empty deposit.
Args:
req (Request): the request holding information to parse
and inject in db
headers (dict): request headers formatted
collection_name (str): the associated client
deposit_id (id): deposit identifier
Returns:
Dictionary of result with the deposit's id, the date
it was completed and no archive.
"""
deposit = Deposit.objects.get(pk=deposit_id)
deposit.complete_date = timezone.now()
deposit.status = DEPOSIT_STATUS_DEPOSITED
deposit.save()
return {
'deposit_id': deposit_id,
'deposit_date': deposit.complete_date,
'status': deposit.status,
'archive': None,
}
def _make_iris(self, req, collection_name, deposit_id):
"""Define the IRI endpoints
Args:
req (Request): The initial request
collection_name (str): client/collection's name
deposit_id (id): Deposit identifier
Returns:
Dictionary of keys with the iris' urls.
"""
args = [collection_name, deposit_id]
return {
iri: req.build_absolute_uri(reverse(iri, args=args))
for iri in [EM_IRI, EDIT_SE_IRI, CONT_FILE_IRI, STATE_IRI]
}
def additional_checks(self, req, headers, collection_name,
deposit_id=None):
"""Permit the child class to enrich additional checks.
Returns:
dict with 'error' detailing the problem.
"""
return {}
def checks(self, req, collection_name, deposit_id=None):
try:
self._collection = DepositCollection.objects.get(
name=collection_name)
except DepositCollection.DoesNotExist:
return make_error_dict(
NOT_FOUND,
'Unknown collection name %s' % collection_name)
username = req.user.username
if username: # unauthenticated request can have the username empty
try:
self._client = DepositClient.objects.get(username=username)
except DepositClient.DoesNotExist:
return make_error_dict(NOT_FOUND,
'Unknown client name %s' % username)
if self._collection.id not in self._client.collections:
return make_error_dict(
FORBIDDEN,
'Client %s cannot access collection %s' % (
username, collection_name))
if deposit_id:
try:
deposit = Deposit.objects.get(pk=deposit_id)
except Deposit.DoesNotExist:
return make_error_dict(
NOT_FOUND,
'Deposit with id %s does not exist' %
deposit_id)
checks = self.restrict_access(req, deposit)
if checks:
return checks
headers = self._read_headers(req)
if headers['on-behalf-of']:
return make_error_dict(MEDIATION_NOT_ALLOWED,
'Mediation is not supported.')
checks = self.additional_checks(req, headers,
collection_name, deposit_id)
if 'error' in checks:
return checks
return {'headers': headers}
def restrict_access(self, req, deposit=None):
if deposit:
if (req.method != 'GET' and
deposit.status != DEPOSIT_STATUS_PARTIAL):
summary = "You can only act on deposit with status '%s'" % (
DEPOSIT_STATUS_PARTIAL, )
description = "This deposit has status '%s'" % deposit.status
return make_error_dict(
BAD_REQUEST, summary=summary,
verbose_description=description)
def _basic_not_allowed_method(self, req, method):
return make_error_response(
req, METHOD_NOT_ALLOWED,
'%s method is not supported on this endpoint' % method)
def get(self, req, *args, **kwargs):
return self._basic_not_allowed_method(req, 'GET')
def post(self, req, *args, **kwargs):
return self._basic_not_allowed_method(req, 'POST')
def put(self, req, *args, **kwargs):
return self._basic_not_allowed_method(req, 'PUT')
def delete(self, req, *args, **kwargs):
return self._basic_not_allowed_method(req, 'DELETE')
class SWHGetDepositAPI(SWHBaseDeposit, metaclass=ABCMeta):
"""Mixin for class to support GET method.
"""
def get(self, req, collection_name, deposit_id, format=None):
"""Endpoint to create/add resources to deposit.
Returns:
200 response when no error during routine occurred
400 if the deposit does not belong to the collection
404 if the deposit or the collection does not exist
"""
checks = self.checks(req, collection_name, deposit_id)
if 'error' in checks:
return make_error_response_from_dict(req, checks['error'])
r = self.process_get(
req, collection_name, deposit_id)
if isinstance(r, tuple):
status, content, content_type = r
return HttpResponse(content,
status=status,
content_type=content_type)
return r
@abstractmethod
def process_get(self, req, collection_name, deposit_id):
"""Routine to deal with the deposit's get processing.
Returns:
Tuple status, stream of content, content-type
"""
pass
class SWHPostDepositAPI(SWHBaseDeposit, metaclass=ABCMeta):
"""Mixin for class to support DELETE method.
"""
def post(self, req, collection_name, deposit_id=None, format=None):
"""Endpoint to create/add resources to deposit.
Returns:
204 response when no error during routine occurred.
400 if the deposit does not belong to the collection
404 if the deposit or the collection does not exist
"""
checks = self.checks(req, collection_name, deposit_id)
if 'error' in checks:
return make_error_response_from_dict(req, checks['error'])
headers = checks['headers']
_status, _iri_key, data = self.process_post(
req, headers, collection_name, deposit_id)
error = data.get('error')
if error:
return make_error_response_from_dict(req, error)
data['packagings'] = ACCEPT_PACKAGINGS
iris = self._make_iris(req, collection_name, data['deposit_id'])
data.update(iris)
response = render(req, 'deposit/deposit_receipt.xml',
context=data,
content_type='application/xml',
status=_status)
response._headers['location'] = 'Location', data[_iri_key]
return response
@abstractmethod
def process_post(self, req, headers, collection_name, deposit_id=None):
"""Routine to deal with the deposit's processing.
Returns
Tuple of:
- response status code (200, 201, etc...)
- key iri (EM_IRI, EDIT_SE_IRI, etc...)
- dictionary of the processing result
"""
pass
class SWHPutDepositAPI(SWHBaseDeposit, metaclass=ABCMeta):
"""Mixin for class to support PUT method.
"""
def put(self, req, collection_name, deposit_id, format=None):
"""Endpoint to update deposit resources.
Returns:
204 response when no error during routine occurred.
400 if the deposit does not belong to the collection
404 if the deposit or the collection does not exist
"""
checks = self.checks(req, collection_name, deposit_id)
if 'error' in checks:
return make_error_response_from_dict(req, checks['error'])
headers = checks['headers']
data = self.process_put(req, headers, collection_name, deposit_id)
error = data.get('error')
if error:
return make_error_response_from_dict(req, error)
return HttpResponse(status=status.HTTP_204_NO_CONTENT)
@abstractmethod
def process_put(self, req, headers, collection_name, deposit_id):
"""Routine to deal with updating a deposit in some way.
Returns
dictionary of the processing result
"""
pass
class SWHDeleteDepositAPI(SWHBaseDeposit, metaclass=ABCMeta):
"""Mixin for class to support DELETE method.
"""
def delete(self, req, collection_name, deposit_id):
"""Endpoint to delete some deposit's resources (archives, deposit).
Returns:
204 response when no error during routine occurred.
400 if the deposit does not belong to the collection
404 if the deposit or the collection does not exist
"""
checks = self.checks(req, collection_name, deposit_id)
if 'error' in checks:
return make_error_response_from_dict(req, checks['error'])
data = self.process_delete(req, collection_name, deposit_id)
error = data.get('error')
if error:
return make_error_response_from_dict(req, error)
return HttpResponse(status=status.HTTP_204_NO_CONTENT)
@abstractmethod
def process_delete(self, req, collection_name, deposit_id):
"""Routine to delete a resource.
This is mostly not allowed except for the
EM_IRI (cf. .api.deposit_update.SWHUpdateArchiveDeposit)
"""
pass
diff --git a/swh/deposit/api/service_document.py b/swh/deposit/api/service_document.py
index 0b04103a..9b79065c 100644
--- a/swh/deposit/api/service_document.py
+++ b/swh/deposit/api/service_document.py
@@ -1,33 +1,33 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# 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
from django.shortcuts import render
-from django.core.urlresolvers import reverse
+from django.urls import reverse
from .common import SWHBaseDeposit, ACCEPT_PACKAGINGS
from .common import ACCEPT_ARCHIVE_CONTENT_TYPES
from ..config import COL_IRI
from ..models import DepositClient, DepositCollection
class SWHServiceDocument(SWHBaseDeposit):
def get(self, req, *args, **kwargs):
client = DepositClient.objects.get(username=req.user)
collections = {}
for col_id in client.collections:
col = DepositCollection.objects.get(pk=col_id)
col_uri = req.build_absolute_uri(reverse(COL_IRI, args=[col.name]))
collections[col.name] = col_uri
context = {
'max_upload_size': self.config['max_upload_size'],
'accept_packagings': ACCEPT_PACKAGINGS,
'accept_content_types': ACCEPT_ARCHIVE_CONTENT_TYPES,
'collections': collections,
}
return render(req, 'deposit/service_document.xml',
context, content_type='application/xml')
diff --git a/swh/deposit/signals.py b/swh/deposit/signals.py
index 83d893a0..13a6739e 100644
--- a/swh/deposit/signals.py
+++ b/swh/deposit/signals.py
@@ -1,83 +1,83 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# 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
"""Module in charge of defining some uncoupled actions on deposit.
Typically, checking that the archives deposited are ok are not
directly testing in the request/answer to avoid too long
computations.
So this is done in the deposit_on_status_ready_for_check callback.
"""
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Deposit
from .config import SWHDefaultConfig, DEPOSIT_STATUS_VERIFIED
from .config import DEPOSIT_STATUS_DEPOSITED
@receiver(post_save, sender=Deposit)
def post_deposit_save(sender, instance, created, raw, using,
update_fields, **kwargs):
"""When a deposit is saved, check for the deposit's status change and
schedule actions accordingly.
When the status passes to deposited, schedule checks.
When the status pass to ready, schedule loading. Otherwise, do
nothing.
Args:
sender (Deposit): The model class
instance (Deposit): The actual instance being saved
created (bool): True if a new record was created
raw (bool): True if the model is saved exactly as presented
(i.e. when loading a fixture). One should not
query/modify other records in the database as the
database might not be in a consistent state yet
using: The database alias being used
update_fields: The set of fields to update as passed to
Model.save(), or None if update_fields wasn’t
passed to save()
"""
default_config = SWHDefaultConfig()
if not default_config.config['checks']:
return
if instance.status not in {DEPOSIT_STATUS_DEPOSITED,
DEPOSIT_STATUS_VERIFIED}:
return
- from django.core.urlresolvers import reverse
+ from django.urls import reverse
from swh.scheduler.utils import create_oneshot_task_dict
args = [instance.collection.name, instance.id]
if instance.status == DEPOSIT_STATUS_DEPOSITED:
# schedule archive check
from swh.deposit.config import PRIVATE_CHECK_DEPOSIT
check_url = reverse(PRIVATE_CHECK_DEPOSIT, args=args)
task = create_oneshot_task_dict(
'swh-deposit-archive-checks',
deposit_check_url=check_url)
else: # instance.status == DEPOSIT_STATUS_VERIFIED:
# schedule loading
from swh.deposit.config import PRIVATE_GET_RAW_CONTENT
from swh.deposit.config import PRIVATE_GET_DEPOSIT_METADATA
from swh.deposit.config import PRIVATE_PUT_DEPOSIT
archive_url = reverse(PRIVATE_GET_RAW_CONTENT, args=args)
meta_url = reverse(PRIVATE_GET_DEPOSIT_METADATA, args=args)
update_url = reverse(PRIVATE_PUT_DEPOSIT, args=args)
task = create_oneshot_task_dict(
'swh-deposit-archive-loading',
archive_url=archive_url,
deposit_meta_url=meta_url,
deposit_update_url=update_url)
default_config.scheduler.create_tasks([task])
diff --git a/swh/deposit/tests/api/test_common.py b/swh/deposit/tests/api/test_common.py
index 74bc0b8b..74479973 100644
--- a/swh/deposit/tests/api/test_common.py
+++ b/swh/deposit/tests/api/test_common.py
@@ -1,39 +1,39 @@
-# Copyright (C) 2017 The Software Heritage developers
+# 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
-from django.core.urlresolvers import reverse
+from django.urls import reverse
from rest_framework import status
from rest_framework.test import APITestCase
from ..common import BasicTestCase, WithAuthTestCase
class IndexNoAuthCase(APITestCase, BasicTestCase):
"""Access to main entry point is ok without authentication
"""
def test_get_home_is_ok(self):
"""Without authentication, endpoint refuses access with 401 response
"""
url = reverse('home')
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn(b'The Software Heritage Deposit', response.content)
class IndexWithAuthCase(WithAuthTestCase, APITestCase, BasicTestCase):
"""Access to main entry point is ok with authentication as well
"""
def test_get_home_is_ok_2(self):
"""Without authentication, endpoint refuses access with 401 response
"""
url = reverse('home')
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn(b'The Software Heritage Deposit', response.content)
diff --git a/swh/deposit/tests/api/test_deposit.py b/swh/deposit/tests/api/test_deposit.py
index fcfac4e9..eb984002 100644
--- a/swh/deposit/tests/api/test_deposit.py
+++ b/swh/deposit/tests/api/test_deposit.py
@@ -1,160 +1,160 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# 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 hashlib
-from django.core.urlresolvers import reverse
+from django.urls import reverse
from io import BytesIO
from rest_framework import status
from rest_framework.test import APITestCase
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_LOAD_SUCCESS
from swh.deposit.config import DEPOSIT_STATUS_LOAD_FAILURE
from swh.deposit.models import Deposit, DepositClient, DepositCollection
from swh.deposit.parsers import parse_xml
from ..common import BasicTestCase, WithAuthTestCase, CommonCreationRoutine
class DepositNoAuthCase(APITestCase, BasicTestCase):
"""Deposit access are protected with basic authentication.
"""
def test_post_will_fail_with_401(self):
"""Without authentication, endpoint refuses access with 401 response
"""
url = reverse(COL_IRI, args=[self.collection.name])
# when
response = self.client.post(url)
# then
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
class DepositFailuresTest(APITestCase, WithAuthTestCase, BasicTestCase,
CommonCreationRoutine):
"""Deposit access are protected with basic authentication.
"""
def setUp(self):
super().setUp()
# Add another user
_collection2 = DepositCollection(name='some')
_collection2.save()
_user = DepositClient.objects.create_user(username='user',
password='user')
_user.collections = [_collection2.id]
self.collection2 = _collection2
def test_access_to_another_user_collection_is_forbidden(self):
"""Access to another user collection should return a 403
"""
url = reverse(COL_IRI, args=[self.collection2.name])
response = self.client.post(url)
self.assertEqual(response.status_code,
status.HTTP_403_FORBIDDEN)
self.assertRegex(response.content.decode('utf-8'),
'Client hal cannot access collection %s' % (
self.collection2.name, ))
def test_delete_on_col_iri_not_supported(self):
"""Delete on col iri should return a 405 response
"""
url = reverse(COL_IRI, args=[self.collection.name])
response = self.client.delete(url)
self.assertEqual(response.status_code,
status.HTTP_405_METHOD_NOT_ALLOWED)
self.assertRegex(response.content.decode('utf-8'),
'DELETE method is not supported on this endpoint')
def create_deposit_with_rejection_status(self):
url = reverse(COL_IRI, args=[self.collection.name])
data = b'some data which is clearly not a zip file'
md5sum = hashlib.md5(data).hexdigest()
external_id = 'some-external-id-1'
# when
response = self.client.post(
url,
content_type='application/zip', # as zip
data=data,
# + headers
CONTENT_LENGTH=len(data),
# other headers needs HTTP_ prefix to be taken into account
HTTP_SLUG=external_id,
HTTP_CONTENT_MD5=md5sum,
HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip',
HTTP_CONTENT_DISPOSITION='attachment; filename=filename0')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
response_content = parse_xml(BytesIO(response.content))
actual_state = response_content['deposit_status']
self.assertEqual(actual_state, DEPOSIT_STATUS_REJECTED)
def test_act_on_deposit_rejected_is_not_permitted(self):
deposit_id = self.create_deposit_with_status(DEPOSIT_STATUS_REJECTED)
deposit = Deposit.objects.get(pk=deposit_id)
assert deposit.status == DEPOSIT_STATUS_REJECTED
response = self.client.post(
reverse(EDIT_SE_IRI, args=[self.collection.name, deposit_id]),
content_type='application/atom+xml;type=entry',
data=self.atom_entry_data1,
HTTP_SLUG='external-id')
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertRegex(
response.content.decode('utf-8'),
"You can only act on deposit with status '%s'" % (
DEPOSIT_STATUS_PARTIAL, ))
def test_add_deposit_with_parent(self):
# given multiple deposit already loaded
deposit_id = self.create_deposit_with_status(
status=DEPOSIT_STATUS_LOAD_SUCCESS,
external_id='some-external-id')
deposit1 = Deposit.objects.get(pk=deposit_id)
self.assertIsNotNone(deposit1)
self.assertEqual(deposit1.external_id, 'some-external-id')
self.assertEqual(deposit1.status, DEPOSIT_STATUS_LOAD_SUCCESS)
deposit_id2 = self.create_deposit_with_status(
status=DEPOSIT_STATUS_LOAD_SUCCESS,
external_id='some-external-id')
deposit2 = Deposit.objects.get(pk=deposit_id2)
self.assertIsNotNone(deposit2)
self.assertEqual(deposit2.external_id, 'some-external-id')
self.assertEqual(deposit2.status, DEPOSIT_STATUS_LOAD_SUCCESS)
deposit_id3 = self.create_deposit_with_status(
status=DEPOSIT_STATUS_LOAD_FAILURE,
external_id='some-external-id')
deposit3 = Deposit.objects.get(pk=deposit_id3)
self.assertIsNotNone(deposit3)
self.assertEqual(deposit3.external_id, 'some-external-id')
self.assertEqual(deposit3.status, DEPOSIT_STATUS_LOAD_FAILURE)
# when
deposit_id3 = self.create_simple_deposit_partial(
external_id='some-external-id')
# then
deposit4 = Deposit.objects.get(pk=deposit_id3)
self.assertIsNotNone(deposit4)
self.assertEqual(deposit4.external_id, 'some-external-id')
self.assertEqual(deposit4.status, DEPOSIT_STATUS_PARTIAL)
self.assertEqual(deposit4.parent, deposit2)
diff --git a/swh/deposit/tests/api/test_deposit_atom.py b/swh/deposit/tests/api/test_deposit_atom.py
index 2f50f050..4220b846 100644
--- a/swh/deposit/tests/api/test_deposit_atom.py
+++ b/swh/deposit/tests/api/test_deposit_atom.py
@@ -1,528 +1,528 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# 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
-from django.core.urlresolvers import reverse
+from django.urls import reverse
from io import BytesIO
from rest_framework import status
from rest_framework.test import APITestCase
from swh.deposit.config import COL_IRI, DEPOSIT_STATUS_DEPOSITED
from swh.deposit.models import Deposit, DepositRequest
from swh.deposit.parsers import parse_xml
from ..common import BasicTestCase, WithAuthTestCase
class DepositAtomEntryTestCase(APITestCase, WithAuthTestCase, BasicTestCase):
"""Try and post atom entry deposit.
"""
def setUp(self):
super().setUp()
self.atom_entry_data0 = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom">
<title>Awesome Compiler</title>
<client>hal</client>
<id>urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a</id>
<external_identifier>%s</external_identifier>
<updated>2017-10-07T15:17:08Z</updated>
<author>some awesome author</author>
<applicationCategory>something</applicationCategory>
<name>awesome-compiler</name>
<description>This is an awesome compiler destined to
awesomely compile stuff
and other stuff</description>
<keywords>compiler,programming,language</keywords>
<dateCreated>2005-10-07T17:17:08Z</dateCreated>
<datePublished>2005-10-07T17:17:08Z</datePublished>
<releaseNotes>release note</releaseNotes>
<relatedLink>related link</relatedLink>
<sponsor></sponsor>
<programmingLanguage>Awesome</programmingLanguage>
<codeRepository>https://hoster.org/awesome-compiler</codeRepository>
<operatingSystem>GNU/Linux</operatingSystem>
<version>0.0.1</version>
<developmentStatus>running</developmentStatus>
<runtimePlatform>all</runtimePlatform>
</entry>"""
self.atom_entry_data1 = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom">
<client>hal</client>
<id>urn:uuid:2225c695-cfb8-4ebb-aaaa-80da344efa6a</id>
<updated>2017-10-07T15:17:08Z</updated>
<author>some awesome author</author>
<applicationCategory>something</applicationCategory>
<name>awesome-compiler</name>
<description>This is an awesome compiler destined to
awesomely compile stuff
and other stuff</description>
<keywords>compiler,programming,language</keywords>
<dateCreated>2005-10-07T17:17:08Z</dateCreated>
<datePublished>2005-10-07T17:17:08Z</datePublished>
<releaseNotes>release note</releaseNotes>
<relatedLink>related link</relatedLink>
<sponsor></sponsor>
<programmingLanguage>Awesome</programmingLanguage>
<codeRepository>https://hoster.org/awesome-compiler</codeRepository>
<operatingSystem>GNU/Linux</operatingSystem>
<version>0.0.1</version>
<developmentStatus>running</developmentStatus>
<runtimePlatform>all</runtimePlatform>
</entry>"""
self.atom_entry_data2 = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom">
<external_identifier>%s</external_identifier>
</entry>"""
self.atom_entry_data_empty_body = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom"></entry>"""
self.atom_entry_data3 = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom">
<something>something</something>
</entry>"""
self.atom_entry_data_atom_only = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom">
<title>Awesome Compiler</title>
<id>urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a</id>
<external_identifier>1785io25c695</external_identifier>
<updated>2017-10-07T15:17:08Z</updated>
<author>some awesome author</author>
</entry>"""
self.atom_entry_data_codemeta = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom"
xmlns:codemeta="https://doi.org/10.5063/SCHEMA/CODEMETA-2.0">
<title>Awesome Compiler</title>
<id>urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a</id>
<external_identifier>1785io25c695</external_identifier>
<codemeta:id>1785io25c695</codemeta:id>
<codemeta:url>origin url</codemeta:url>
<codemeta:identifier>other identifier, DOI, ARK</codemeta:identifier>
<codemeta:applicationCategory>Domain</codemeta:applicationCategory>
<codemeta:description>description</codemeta:description>
<codemeta:keywords>key-word 1</codemeta:keywords>
<codemeta:keywords>key-word 2</codemeta:keywords>
<codemeta:dateCreated>creation date</codemeta:dateCreated>
<codemeta:datePublished>publication date</codemeta:datePublished>
<codemeta:releaseNotes>comment</codemeta:releaseNotes>
<codemeta:referencePublication>
<codemeta:name> article name</codemeta:name>
<codemeta:identifier> article id </codemeta:identifier>
</codemeta:referencePublication>
<codemeta:isPartOf>
<codemeta:type> Collaboration/Projet </codemeta:type>
<codemeta:name> project name</codemeta:name>
<codemeta:identifier> id </codemeta:identifier>
</codemeta:isPartOf>
<codemeta:relatedLink>see also </codemeta:relatedLink>
<codemeta:funding>Sponsor A </codemeta:funding>
<codemeta:funding>Sponsor B</codemeta:funding>
<codemeta:operatingSystem>Platform/OS </codemeta:operatingSystem>
<codemeta:softwareRequirements>dependencies </codemeta:softwareRequirements>
<codemeta:softwareVersion>Version</codemeta:softwareVersion>
<codemeta:developmentStatus>active </codemeta:developmentStatus>
<codemeta:license>
<codemeta:name>license</codemeta:name>
<codemeta:url>url spdx</codemeta:url>
</codemeta:license>
<codemeta:runtimePlatform>.Net Framework 3.0 </codemeta:runtimePlatform>
<codemeta:runtimePlatform>Python2.3</codemeta:runtimePlatform>
<codemeta:author>
<codemeta:name> author1 </codemeta:name>
<codemeta:affiliation> Inria </codemeta:affiliation>
<codemeta:affiliation> UPMC </codemeta:affiliation>
</codemeta:author>
<codemeta:author>
<codemeta:name> author2 </codemeta:name>
<codemeta:affiliation> Inria </codemeta:affiliation>
<codemeta:affiliation> UPMC </codemeta:affiliation>
</codemeta:author>
<codemeta:codeRepository>http://code.com</codemeta:codeRepository>
<codemeta:programmingLanguage>language 1</codemeta:programmingLanguage>
<codemeta:programmingLanguage>language 2</codemeta:programmingLanguage>
<codemeta:issueTracker>http://issuetracker.com</codemeta:issueTracker>
</entry>""" # noqa
self.atom_entry_data_dc_codemeta = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom"
xmlns:dcterms="http://purl.org/dc/terms/"
xmlns:codemeta="https://doi.org/10.5063/SCHEMA/CODEMETA-2.0">
<external_identifier>%s</external_identifier>
<dcterms:identifier>hal-01587361</dcterms:identifier>
<dcterms:identifier>https://hal.inria.fr/hal-01587361</dcterms:identifier>
<dcterms:identifier>https://hal.inria.fr/hal-01587361/document</dcterms:identifier>
<dcterms:identifier>https://hal.inria.fr/hal-01587361/file/AffectationRO-v1.0.0.zip</dcterms:identifier>
<dcterms:identifier>doi:10.5281/zenodo.438684</dcterms:identifier>
<dcterms:title xml:lang="en">The assignment problem</dcterms:title>
<dcterms:title xml:lang="fr">AffectationRO</dcterms:title>
<dcterms:creator>Gruenpeter, Morane</dcterms:creator>
<dcterms:subject>[INFO] Computer Science [cs]</dcterms:subject>
<dcterms:subject>[INFO.INFO-RO] Computer Science [cs]/Operations Research [cs.RO]</dcterms:subject>
<dcterms:type>SOFTWARE</dcterms:type>
<dcterms:abstract xml:lang="en">Project in OR: The assignment problemA java implementation for the assignment problem first release</dcterms:abstract>
<dcterms:abstract xml:lang="fr">description fr</dcterms:abstract>
<dcterms:created>2015-06-01</dcterms:created>
<dcterms:available>2017-10-19</dcterms:available>
<dcterms:language>en</dcterms:language>
<codemeta:url>url stable</codemeta:url>
<codemeta:version>Version sur hal </codemeta:version>
<codemeta:softwareVersion>Version entre par lutilisateur</codemeta:softwareVersion>
<codemeta:keywords>Mots-cls</codemeta:keywords>
<codemeta:releaseNotes>Commentaire</codemeta:releaseNotes>
<codemeta:referencePublication>Rfrence interne </codemeta:referencePublication>
<codemeta:isPartOf>
<codemeta:type> Collaboration/Projet </codemeta:type>
<codemeta:name> nom du projet</codemeta:name>
<codemeta:identifier> id </codemeta:identifier>
</codemeta:isPartOf>
<codemeta:relatedLink>Voir aussi </codemeta:relatedLink>
<codemeta:funding>Financement </codemeta:funding>
<codemeta:funding>Projet ANR </codemeta:funding>
<codemeta:funding>Projet Europen </codemeta:funding>
<codemeta:operatingSystem>Platform/OS </codemeta:operatingSystem>
<codemeta:softwareRequirements>Dpendances </codemeta:softwareRequirements>
<codemeta:developmentStatus>Etat du dveloppement </codemeta:developmentStatus>
<codemeta:license>
<codemeta:name>license</codemeta:name>
<codemeta:url>url spdx</codemeta:url>
</codemeta:license>
<codemeta:runtimePlatform>Outils de dveloppement- outil no1 </codemeta:runtimePlatform>
<codemeta:runtimePlatform>Outils de dveloppement- outil no2 </codemeta:runtimePlatform>
<codemeta:codeRepository>http://code.com</codemeta:codeRepository>
<codemeta:programmingLanguage>language 1</codemeta:programmingLanguage>
<codemeta:programmingLanguage>language 2</codemeta:programmingLanguage>
</entry>""" # noqa
self.atom_entry_tei = b"""<TEI><teiHeader><fileDesc><titleStmt><title>HAL TEI export of hal-01587083</title></titleStmt><publicationStmt><distributor>CCSD</distributor><availability status="restricted"><licence target="http://creativecommons.org/licenses/by/4.0/">Distributed under a Creative Commons Attribution 4.0 International License</licence></availability><date when="2017-10-03T17:21:03+02:00"/></publicationStmt><sourceDesc><p part="N">HAL API platform</p></sourceDesc></fileDesc></teiHeader><text><body><listBibl><biblFull><titleStmt><title xml:lang="en">questionnaire software metadata</title><author role="aut"><persName><forename type="first">Morane</forename><surname>Gruenpeter</surname></persName><email type="md5">7de56c632362954fa84172cad80afe4e</email><email type="domain">inria.fr</email><ptr type="url" target="moranegg.github.io"/><idno type="halauthorid">1556733</idno><affiliation ref="#struct-474639"/></author><editor role="depositor"><persName><forename>Morane</forename><surname>Gruenpeter</surname></persName><email type="md5">f85a43a5fb4a2e0778a77e017f28c8fd</email><email type="domain">gmail.com</email></editor></titleStmt><editionStmt><edition n="v1" type="current"><date type="whenSubmitted">2017-09-29 11:21:32</date><date type="whenModified">2017-10-03 17:20:13</date><date type="whenReleased">2017-10-03 17:20:13</date><date type="whenProduced">2017-09-29</date><date type="whenEndEmbargoed">2017-09-29</date><ref type="file" target="https://hal.inria.fr/hal-01587083/document"><date notBefore="2017-09-29"/></ref><ref type="file" subtype="author" n="1" target="https://hal.inria.fr/hal-01587083/file/questionnaire.zip"><date notBefore="2017-09-29"/></ref></edition><respStmt><resp>contributor</resp><name key="442239"><persName><forename>Morane</forename><surname>Gruenpeter</surname></persName><email type="md5">f85a43a5fb4a2e0778a77e017f28c8fd</email><email type="domain">gmail.com</email></name></respStmt></editionStmt><publicationStmt><distributor>CCSD</distributor><idno type="halId">hal-01587083</idno><idno type="halUri">https://hal.inria.fr/hal-01587083</idno><idno type="halBibtex">gruenpeter:hal-01587083</idno><idno type="halRefHtml">2017</idno><idno type="halRef">2017</idno></publicationStmt><seriesStmt/><notesStmt/><sourceDesc><biblStruct><analytic><title xml:lang="en">questionnaire software metadata</title><author role="aut"><persName><forename type="first">Morane</forename><surname>Gruenpeter</surname></persName><email type="md5">7de56c632362954fa84172cad80afe4e</email><email type="domain">inria.fr</email><ptr type="url" target="moranegg.github.io"/><idno type="halauthorid">1556733</idno><affiliation ref="#struct-474639"/></author></analytic><monogr><imprint/></monogr></biblStruct></sourceDesc><profileDesc><langUsage><language ident="en">English</language></langUsage><textClass><classCode scheme="halDomain" n="info">Computer Science [cs]</classCode><classCode scheme="halTypology" n="SOFTWARE">Software</classCode></textClass></profileDesc></biblFull></listBibl></body><back><listOrg type="structures"><org type="laboratory" xml:id="struct-474639" status="VALID"><orgName>IRILL</orgName><orgName type="acronym">Initiative pour la Recherche et l'Innovation sur le Logiciel Libre</orgName><desc><address><country key="FR"/></address><ref type="url">https://www.irill.org/</ref></desc><listRelation><relation active="#struct-93591" type="direct"/><relation active="#struct-300009" type="direct"/><relation active="#struct-300301" type="direct"/></listRelation></org><org type="institution" xml:id="struct-93591" status="VALID"><orgName>Universite Pierre et Marie Curie - Paris 6</orgName><orgName type="acronym">UPMC</orgName><desc><address><addrLine>4 place Jussieu - 75005 Paris</addrLine><country key="FR"/></address><ref type="url">http://www.upmc.fr/</ref></desc></org><org type="institution" xml:id="struct-300009" status="VALID"><orgName>Institut National de Recherche en Informatique et en Automatique</orgName><orgName type="acronym">Inria</orgName><desc><address><addrLine>Domaine de VoluceauRocquencourt - BP 10578153 Le Chesnay Cedex</addrLine><country key="FR"/></address><ref type="url">http://www.inria.fr/en/</ref></desc></org><org type="institution" xml:id="struct-300301" status="VALID"><orgName>Universite Paris Diderot - Paris 7</orgName><orgName type="acronym">UPD7</orgName><desc><address><addrLine>5 rue Thomas-Mann - 75205 Paris cedex 13</addrLine><country key="FR"/></address><ref type="url">http://www.univ-paris-diderot.fr</ref></desc></org></listOrg></back></text></TEI>""" # noqa
self.atom_entry_data_badly_formatted = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom"</entry>"""
self.atom_error_with_decimal = b"""<?xml version="1.0" encoding="utf-8"?>
<entry xmlns="http://www.w3.org/2005/Atom" xmlns:codemeta="https://doi.org/10.5063/SCHEMA/CODEMETA-2.0">
<title>Composing a Web of Audio Applications</title>
<client>hal</client>
<id>hal-01243065</id>
<external_identifier>hal-01243065</external_identifier>
<codemeta:url>https://hal-test.archives-ouvertes.fr/hal-01243065</codemeta:url>
<codemeta:applicationCategory>test</codemeta:applicationCategory>
<codemeta:name/>
<description/>
<codemeta:keywords>DSP programming,Web,Composability,Faust</codemeta:keywords>
<codemeta:dateCreated>2017-05-03T16:08:47+02:00</codemeta:dateCreated>
<codemeta:description>The Web offers a great opportunity to share, deploy and use programs without installation difficulties. In this article we explore the idea of freely combining/composing real-time audio applications deployed on the Web using Faust audio DSP language.</codemeta:description>
<codemeta:version>1</codemeta:version>
<codemeta:softwareVersion>10.4</codemeta:softwareVersion>
<codemeta:runtimePlatform>phpstorm</codemeta:runtimePlatform>
<codemeta:developmentStatus>stable</codemeta:developmentStatus>
<codeRepository/>
<platform>linux</platform>
<codemeta:programmingLanguage>php</codemeta:programmingLanguage>
<codemeta:programmingLanguage>python</codemeta:programmingLanguage>
<codemeta:programmingLanguage>C</codemeta:programmingLanguage>
<codemeta:license>
<codemeta:name>GNU General Public License v3.0 only</codemeta:name>
</codemeta:license>
<codemeta:license>
<codemeta:name>CeCILL Free Software License Agreement v1.1</codemeta:name>
</codemeta:license>
<author>
<name>HAL</name>
<email>hal@ccsd.cnrs.fr</email>
</author>
<contributor>
<name>Someone Nice</name>
<email>someone@nice.fr</email>
<codemeta:affiliation>FFJ</codemeta:affiliation>
</contributor>
</entry>
""" # noqa
def test_post_deposit_atom_entry_serialization_error(self):
"""Posting an initial atom entry should return 201 with deposit receipt
"""
# given
# when
response = self.client.post(
reverse(COL_IRI, args=[self.collection.name]),
content_type='application/atom+xml;type=entry',
data=self.atom_error_with_decimal,
HTTP_SLUG='external-id',
HTTP_IN_PROGRESS='false')
# 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)
dr = DepositRequest.objects.get(deposit=deposit)
self.assertIsNotNone(dr.metadata)
sw_version = dr.metadata.get('codemeta:softwareVersion')
self.assertEqual(sw_version, '10.4')
def test_post_deposit_atom_empty_body_request(self):
"""Posting empty body request should return a 400 response
"""
response = self.client.post(
reverse(COL_IRI, args=[self.collection.name]),
content_type='application/atom+xml;type=entry',
data=self.atom_entry_data_empty_body)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
def test_post_deposit_atom_badly_formatted_is_a_bad_request(self):
"""Posting a badly formatted atom should return a 400 response
"""
response = self.client.post(
reverse(COL_IRI, args=[self.collection.name]),
content_type='application/atom+xml;type=entry',
data=self.atom_entry_data_badly_formatted)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
def test_post_deposit_atom_without_slug_header_is_bad_request(self):
"""Posting an atom entry without a slug header should return a 400
"""
url = reverse(COL_IRI, args=[self.collection.name])
# when
response = self.client.post(
url,
content_type='application/atom+xml;type=entry',
data=self.atom_entry_data0,
# + headers
HTTP_IN_PROGRESS='false')
self.assertIn(b'Missing SLUG header', response.content)
self.assertEqual(response.status_code,
status.HTTP_400_BAD_REQUEST)
def test_post_deposit_atom_unknown_collection(self):
"""Posting an atom entry to an unknown collection should return a 404
"""
response = self.client.post(
reverse(COL_IRI, args=['unknown-one']),
content_type='application/atom+xml;type=entry',
data=self.atom_entry_data3,
HTTP_SLUG='something')
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
def test_post_deposit_atom_entry_initial(self):
"""Posting an initial atom entry should return 201 with deposit receipt
"""
# given
external_id = 'urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a'
with self.assertRaises(Deposit.DoesNotExist):
Deposit.objects.get(external_id=external_id)
atom_entry_data = self.atom_entry_data0 % external_id.encode('utf-8')
# when
response = self.client.post(
reverse(COL_IRI, args=[self.collection.name]),
content_type='application/atom+xml;type=entry',
data=atom_entry_data,
HTTP_SLUG='external-id',
HTTP_IN_PROGRESS='false')
# 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.collection, self.collection)
self.assertEqual(deposit.external_id, external_id)
self.assertEqual(deposit.status, DEPOSIT_STATUS_DEPOSITED)
self.assertEqual(deposit.client, self.user)
# one associated request to a deposit
deposit_request = DepositRequest.objects.get(deposit=deposit)
self.assertIsNotNone(deposit_request.metadata)
self.assertEqual(
deposit_request.raw_metadata, atom_entry_data.decode('utf-8'))
self.assertFalse(bool(deposit_request.archive))
def test_post_deposit_atom_entry_with_codemeta(self):
"""Posting an initial atom entry should return 201 with deposit receipt
"""
# given
external_id = 'urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a'
with self.assertRaises(Deposit.DoesNotExist):
Deposit.objects.get(external_id=external_id)
atom_entry_data = self.atom_entry_data_dc_codemeta % (
external_id.encode('utf-8'), )
# when
response = self.client.post(
reverse(COL_IRI, args=[self.collection.name]),
content_type='application/atom+xml;type=entry',
data=atom_entry_data,
HTTP_SLUG='external-id',
HTTP_IN_PROGRESS='false')
# 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.collection, self.collection)
self.assertEqual(deposit.external_id, external_id)
self.assertEqual(deposit.status, DEPOSIT_STATUS_DEPOSITED)
self.assertEqual(deposit.client, self.user)
# one associated request to a deposit
deposit_request = DepositRequest.objects.get(deposit=deposit)
self.assertIsNotNone(deposit_request.metadata)
self.assertEqual(
deposit_request.raw_metadata, atom_entry_data.decode('utf-8'))
self.assertFalse(bool(deposit_request.archive))
def test_post_deposit_atom_entry_tei(self):
"""Posting initial atom entry as TEI should return 201 with receipt
"""
# given
external_id = 'urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a'
with self.assertRaises(Deposit.DoesNotExist):
Deposit.objects.get(external_id=external_id)
atom_entry_data = self.atom_entry_tei
# when
response = self.client.post(
reverse(COL_IRI, args=[self.collection.name]),
content_type='application/atom+xml;type=entry',
data=atom_entry_data,
HTTP_SLUG=external_id,
HTTP_IN_PROGRESS='false')
# 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.collection, self.collection)
self.assertEqual(deposit.external_id, external_id)
self.assertEqual(deposit.status, DEPOSIT_STATUS_DEPOSITED)
self.assertEqual(deposit.client, self.user)
# one associated request to a deposit
deposit_request = DepositRequest.objects.get(deposit=deposit)
self.assertIsNotNone(deposit_request.metadata)
self.assertEqual(
deposit_request.raw_metadata, atom_entry_data.decode('utf-8'))
self.assertFalse(bool(deposit_request.archive))
def test_post_deposit_atom_entry_multiple_steps(self):
"""After initial deposit, updating a deposit should return a 201
"""
# given
external_id = 'urn:uuid:2225c695-cfb8-4ebb-aaaa-80da344efa6a'
with self.assertRaises(Deposit.DoesNotExist):
deposit = Deposit.objects.get(external_id=external_id)
# when
response = self.client.post(
reverse(COL_IRI, args=[self.collection.name]),
content_type='application/atom+xml;type=entry',
data=self.atom_entry_data1,
HTTP_IN_PROGRESS='True',
HTTP_SLUG=external_id)
# then
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
response_content = parse_xml(BytesIO(response.content))
deposit_id = int(response_content['deposit_id'])
deposit = Deposit.objects.get(pk=deposit_id)
self.assertEqual(deposit.collection, self.collection)
self.assertEqual(deposit.external_id, external_id)
self.assertEqual(deposit.status, 'partial')
self.assertEqual(deposit.client, self.user)
# one associated request to a deposit
deposit_requests = DepositRequest.objects.filter(deposit=deposit)
self.assertEqual(len(deposit_requests), 1)
atom_entry_data = self.atom_entry_data2 % external_id.encode('utf-8')
update_uri = response._headers['location'][1]
# when updating the first deposit post
response = self.client.post(
update_uri,
content_type='application/atom+xml;type=entry',
data=atom_entry_data,
HTTP_IN_PROGRESS='False')
# then
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
response_content = parse_xml(BytesIO(response.content))
deposit_id = int(response_content['deposit_id'])
deposit = Deposit.objects.get(pk=deposit_id)
self.assertEqual(deposit.collection, self.collection)
self.assertEqual(deposit.external_id, external_id)
self.assertEqual(deposit.status, DEPOSIT_STATUS_DEPOSITED)
self.assertEqual(deposit.client, self.user)
self.assertEqual(len(Deposit.objects.all()), 1)
# now 2 associated requests to a same deposit
deposit_requests = DepositRequest.objects.filter(
deposit=deposit).order_by('id')
self.assertEqual(len(deposit_requests), 2)
expected_meta = [
{
'metadata': parse_xml(self.atom_entry_data1),
'raw_metadata': self.atom_entry_data1.decode('utf-8'),
},
{
'metadata': parse_xml(atom_entry_data),
'raw_metadata': atom_entry_data.decode('utf-8'),
}
]
for i, deposit_request in enumerate(deposit_requests):
actual_metadata = deposit_request.metadata
self.assertEqual(actual_metadata,
expected_meta[i]['metadata'])
self.assertEqual(deposit_request.raw_metadata,
expected_meta[i]['raw_metadata'])
self.assertFalse(bool(deposit_request.archive))
diff --git a/swh/deposit/tests/api/test_deposit_binary.py b/swh/deposit/tests/api/test_deposit_binary.py
index f8e60b7d..59aeb15c 100644
--- a/swh/deposit/tests/api/test_deposit_binary.py
+++ b/swh/deposit/tests/api/test_deposit_binary.py
@@ -1,645 +1,645 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# 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
from django.core.files.uploadedfile import InMemoryUploadedFile
-from django.core.urlresolvers import reverse
+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.models import Deposit, DepositRequest
from swh.deposit.parsers import parse_xml
from ..common import (
BasicTestCase, WithAuthTestCase, create_arborescence_archive,
FileSystemCreationRoutine
)
class DepositTestCase(APITestCase, WithAuthTestCase, BasicTestCase,
FileSystemCreationRoutine):
"""Try and upload one single deposit
"""
def setUp(self):
super().setUp()
self.atom_entry_data0 = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom">
<title>Awesome Compiler</title>
<client>hal</client>
<id>urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a</id>
<external_identifier>%s</external_identifier>
<updated>2017-10-07T15:17:08Z</updated>
<author>some awesome author</author>
<applicationCategory>something</applicationCategory>
<name>awesome-compiler</name>
<description>This is an awesome compiler destined to
awesomely compile stuff
and other stuff</description>
<keywords>compiler,programming,language</keywords>
<dateCreated>2005-10-07T17:17:08Z</dateCreated>
<datePublished>2005-10-07T17:17:08Z</datePublished>
<releaseNotes>release note</releaseNotes>
<relatedLink>related link</relatedLink>
<sponsor></sponsor>
<programmingLanguage>Awesome</programmingLanguage>
<codeRepository>https://hoster.org/awesome-compiler</codeRepository>
<operatingSystem>GNU/Linux</operatingSystem>
<version>0.0.1</version>
<developmentStatus>running</developmentStatus>
<runtimePlatform>all</runtimePlatform>
</entry>"""
self.atom_entry_data1 = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom">
<client>hal</client>
<id>urn:uuid:2225c695-cfb8-4ebb-aaaa-80da344efa6a</id>
<updated>2017-10-07T15:17:08Z</updated>
<author>some awesome author</author>
<applicationCategory>something</applicationCategory>
<name>awesome-compiler</name>
<description>This is an awesome compiler destined to
awesomely compile stuff
and other stuff</description>
<keywords>compiler,programming,language</keywords>
<dateCreated>2005-10-07T17:17:08Z</dateCreated>
<datePublished>2005-10-07T17:17:08Z</datePublished>
<releaseNotes>release note</releaseNotes>
<relatedLink>related link</relatedLink>
<sponsor></sponsor>
<programmingLanguage>Awesome</programmingLanguage>
<codeRepository>https://hoster.org/awesome-compiler</codeRepository>
<operatingSystem>GNU/Linux</operatingSystem>
<version>0.0.1</version>
<developmentStatus>running</developmentStatus>
<runtimePlatform>all</runtimePlatform>
</entry>"""
self.atom_entry_data2 = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom">
<external_identifier>%s</external_identifier>
</entry>"""
self.atom_entry_data_empty_body = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom"></entry>"""
self.atom_entry_data3 = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom">
<something>something</something>
</entry>"""
self.data_atom_entry_ok = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom"
xmlns:dcterms="http://purl.org/dc/terms/">
<title>Title</title>
<id>urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a</id>
<updated>2005-10-07T17:17:08Z</updated>
<author><name>Contributor</name></author>
<summary type="text">The abstract</summary>
<!-- some embedded metadata -->
<dcterms:abstract>The abstract</dcterms:abstract>
<dcterms:accessRights>Access Rights</dcterms:accessRights>
<dcterms:alternative>Alternative Title</dcterms:alternative>
<dcterms:available>Date Available</dcterms:available>
<dcterms:bibliographicCitation>Bibliographic Citation</dcterms:bibliographicCitation> # noqa
<dcterms:contributor>Contributor</dcterms:contributor>
<dcterms:description>Description</dcterms:description>
<dcterms:hasPart>Has Part</dcterms:hasPart>
<dcterms:hasVersion>Has Version</dcterms:hasVersion>
<dcterms:identifier>Identifier</dcterms:identifier>
<dcterms:isPartOf>Is Part Of</dcterms:isPartOf>
<dcterms:publisher>Publisher</dcterms:publisher>
<dcterms:references>References</dcterms:references>
<dcterms:rightsHolder>Rights Holder</dcterms:rightsHolder>
<dcterms:source>Source</dcterms:source>
<dcterms:title>Title</dcterms:title>
<dcterms:type>Type</dcterms:type>
</entry>"""
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)
diff --git a/swh/deposit/tests/api/test_deposit_check.py b/swh/deposit/tests/api/test_deposit_check.py
index 1b9c5d2d..f0585d88 100644
--- a/swh/deposit/tests/api/test_deposit_check.py
+++ b/swh/deposit/tests/api/test_deposit_check.py
@@ -1,236 +1,236 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# 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 unittest
-from django.core.urlresolvers import reverse
+from django.urls import reverse
import pytest
from rest_framework import status
from rest_framework.test import APITestCase
from swh.deposit.config import (
DEPOSIT_STATUS_VERIFIED, PRIVATE_CHECK_DEPOSIT,
DEPOSIT_STATUS_DEPOSITED, DEPOSIT_STATUS_REJECTED
)
from swh.deposit.api.private.deposit_check import (
SWHChecksDeposit, MANDATORY_ARCHIVE_INVALID,
MANDATORY_FIELDS_MISSING, INCOMPATIBLE_URL_FIELDS,
MANDATORY_ARCHIVE_UNSUPPORTED, ALTERNATE_FIELDS_MISSING,
MANDATORY_ARCHIVE_MISSING
)
from swh.deposit.models import Deposit
from ..common import BasicTestCase, WithAuthTestCase, CommonCreationRoutine
from ..common import FileSystemCreationRoutine
@pytest.mark.fs
class CheckDepositTest(APITestCase, WithAuthTestCase,
BasicTestCase, CommonCreationRoutine,
FileSystemCreationRoutine):
"""Check deposit endpoints.
"""
def setUp(self):
super().setUp()
def test_deposit_ok(self):
"""Proper deposit should succeed the checks (-> status ready)
"""
deposit_id = self.create_simple_binary_deposit(status_partial=True)
deposit_id = self.update_binary_deposit(deposit_id,
status_partial=False)
deposit = Deposit.objects.get(pk=deposit_id)
self.assertEqual(deposit.status, DEPOSIT_STATUS_DEPOSITED)
url = reverse(PRIVATE_CHECK_DEPOSIT,
args=[self.collection.name, deposit.id])
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data['status'], DEPOSIT_STATUS_VERIFIED)
deposit = Deposit.objects.get(pk=deposit.id)
self.assertEqual(deposit.status, DEPOSIT_STATUS_VERIFIED)
def test_deposit_invalid_tarball(self):
"""Deposit with tarball (of 1 tarball) should fail the checks: rejected
"""
for archive_extension in ['zip', 'tar', 'tar.gz', 'tar.bz2', 'tar.xz']:
deposit_id = self.create_deposit_archive_with_archive(
archive_extension)
deposit = Deposit.objects.get(pk=deposit_id)
self.assertEqual(DEPOSIT_STATUS_DEPOSITED, deposit.status)
url = reverse(PRIVATE_CHECK_DEPOSIT,
args=[self.collection.name, deposit.id])
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data['status'], DEPOSIT_STATUS_REJECTED)
details = data['details']
# archive checks failure
self.assertEqual(len(details['archive']), 1)
self.assertEqual(details['archive'][0]['summary'],
MANDATORY_ARCHIVE_INVALID)
deposit = Deposit.objects.get(pk=deposit.id)
self.assertEqual(deposit.status, DEPOSIT_STATUS_REJECTED)
def test_deposit_ko_missing_tarball(self):
"""Deposit without archive should fail the checks: rejected
"""
deposit_id = self.create_deposit_ready() # no archive, only atom
deposit = Deposit.objects.get(pk=deposit_id)
self.assertEqual(DEPOSIT_STATUS_DEPOSITED, deposit.status)
url = reverse(PRIVATE_CHECK_DEPOSIT,
args=[self.collection.name, deposit.id])
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data['status'], DEPOSIT_STATUS_REJECTED)
details = data['details']
# archive checks failure
self.assertEqual(len(details['archive']), 1)
self.assertEqual(details['archive'][0]['summary'],
MANDATORY_ARCHIVE_MISSING)
deposit = Deposit.objects.get(pk=deposit.id)
self.assertEqual(deposit.status, DEPOSIT_STATUS_REJECTED)
def test_deposit_ko_unsupported_tarball(self):
"""Deposit with an unsupported tarball should fail the checks: rejected
"""
deposit_id = self.create_deposit_with_invalid_archive()
deposit = Deposit.objects.get(pk=deposit_id)
self.assertEqual(DEPOSIT_STATUS_DEPOSITED, deposit.status)
url = reverse(PRIVATE_CHECK_DEPOSIT,
args=[self.collection.name, deposit.id])
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data['status'], DEPOSIT_STATUS_REJECTED)
details = data['details']
# archive checks failure
self.assertEqual(len(details['archive']), 1)
self.assertEqual(details['archive'][0]['summary'],
MANDATORY_ARCHIVE_UNSUPPORTED)
# metadata check failure
self.assertEqual(len(details['metadata']), 2)
mandatory = details['metadata'][0]
self.assertEqual(mandatory['summary'], MANDATORY_FIELDS_MISSING)
self.assertEqual(set(mandatory['fields']),
set(['url', 'external_identifier', 'author']))
alternate = details['metadata'][1]
self.assertEqual(alternate['summary'], ALTERNATE_FIELDS_MISSING)
self.assertEqual(alternate['fields'], ['name or title'])
# url check failure
self.assertEqual(details['url']['summary'], INCOMPATIBLE_URL_FIELDS)
deposit = Deposit.objects.get(pk=deposit.id)
self.assertEqual(deposit.status, DEPOSIT_STATUS_REJECTED)
def test_check_deposit_metadata_ok(self):
"""Proper deposit should succeed the checks (-> status ready)
with all **MUST** metadata
using the codemeta metadata test set
"""
deposit_id = self.create_simple_binary_deposit(status_partial=True)
deposit_id_metadata = self.add_metadata_to_deposit(deposit_id)
self.assertEqual(deposit_id, deposit_id_metadata)
deposit = Deposit.objects.get(pk=deposit_id)
self.assertEqual(deposit.status, DEPOSIT_STATUS_DEPOSITED)
url = reverse(PRIVATE_CHECK_DEPOSIT,
args=[self.collection.name, deposit.id])
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data['status'], DEPOSIT_STATUS_VERIFIED)
deposit = Deposit.objects.get(pk=deposit.id)
self.assertEqual(deposit.status, DEPOSIT_STATUS_VERIFIED)
class CheckMetadata(unittest.TestCase, SWHChecksDeposit):
def test_check_metadata_ok(self):
actual_check, detail = self._check_metadata({
'url': 'something',
'external_identifier': 'something-else',
'name': 'foo',
'author': 'someone',
})
self.assertTrue(actual_check)
self.assertIsNone(detail)
def test_check_metadata_ok2(self):
actual_check, detail = self._check_metadata({
'url': 'something',
'external_identifier': 'something-else',
'title': 'bar',
'author': 'someone',
})
self.assertTrue(actual_check)
self.assertIsNone(detail)
def test_check_metadata_ko(self):
"""Missing optional field should be caught
"""
actual_check, error_detail = self._check_metadata({
'url': 'something',
'external_identifier': 'something-else',
'author': 'someone',
})
expected_error = {
'metadata': [{
'summary': 'Mandatory alternate fields are missing',
'fields': ['name or title'],
}]
}
self.assertFalse(actual_check)
self.assertEqual(error_detail, expected_error)
def test_check_metadata_ko2(self):
"""Missing mandatory fields should be caught
"""
actual_check, error_detail = self._check_metadata({
'url': 'something',
'external_identifier': 'something-else',
'title': 'foobar',
})
expected_error = {
'metadata': [{
'summary': 'Mandatory fields are missing',
'fields': ['author'],
}]
}
self.assertFalse(actual_check)
self.assertEqual(error_detail, expected_error)
diff --git a/swh/deposit/tests/api/test_deposit_delete.py b/swh/deposit/tests/api/test_deposit_delete.py
index 847dd0e6..806d6325 100644
--- a/swh/deposit/tests/api/test_deposit_delete.py
+++ b/swh/deposit/tests/api/test_deposit_delete.py
@@ -1,113 +1,113 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# 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
-from django.core.urlresolvers import reverse
+from django.urls import reverse
from rest_framework import status
from rest_framework.test import APITestCase
from swh.deposit.config import EDIT_SE_IRI, EM_IRI, ARCHIVE_KEY, METADATA_KEY
from swh.deposit.config import DEPOSIT_STATUS_DEPOSITED
from swh.deposit.models import Deposit, DepositRequest
from ..common import BasicTestCase, WithAuthTestCase, CommonCreationRoutine
class DepositDeleteTest(APITestCase, WithAuthTestCase, BasicTestCase,
CommonCreationRoutine):
def test_delete_archive_on_partial_deposit_works(self):
"""Removing partial deposit's archive should return a 204 response
"""
# given
deposit_id = self.create_deposit_partial()
deposit = Deposit.objects.get(pk=deposit_id)
deposit_requests = DepositRequest.objects.filter(deposit=deposit)
self.assertEqual(len(deposit_requests), 2)
for dr in deposit_requests:
if dr.type == ARCHIVE_KEY:
continue
elif dr.type == METADATA_KEY:
continue
else:
self.fail('only archive and metadata type should exist '
'in this test context')
# when
update_uri = reverse(EM_IRI, args=[self.collection.name, deposit_id])
response = self.client.delete(update_uri)
# then
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
deposit = Deposit.objects.get(pk=deposit_id)
requests = list(DepositRequest.objects.filter(deposit=deposit))
self.assertEqual(len(requests), 2)
self.assertEqual(requests[0].type, 'metadata')
self.assertEqual(requests[1].type, 'metadata')
def test_delete_archive_on_undefined_deposit_fails(self):
"""Delete undefined deposit returns a 404 response
"""
# when
update_uri = reverse(EM_IRI, args=[self.collection.name, 999])
response = self.client.delete(update_uri)
# then
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
def test_delete_archive_on_non_partial_deposit_fails(self):
"""Delete !partial status deposit should return a 400 response"""
deposit_id = self.create_deposit_ready()
deposit = Deposit.objects.get(pk=deposit_id)
self.assertEqual(deposit.status, DEPOSIT_STATUS_DEPOSITED)
# when
update_uri = reverse(EM_IRI, args=[self.collection.name, deposit_id])
response = self.client.delete(update_uri)
# then
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
deposit = Deposit.objects.get(pk=deposit_id)
self.assertIsNotNone(deposit)
def test_delete_partial_deposit_works(self):
"""Delete deposit should return a 204 response
"""
# given
deposit_id = self.create_simple_deposit_partial()
deposit = Deposit.objects.get(pk=deposit_id)
assert deposit.id == deposit_id
# when
url = reverse(EDIT_SE_IRI, args=[self.collection.name, deposit_id])
response = self.client.delete(url)
# then
self.assertEqual(response.status_code,
status.HTTP_204_NO_CONTENT)
deposit_requests = list(DepositRequest.objects.filter(deposit=deposit))
self.assertEqual(deposit_requests, [])
deposits = list(Deposit.objects.filter(pk=deposit_id))
self.assertEqual(deposits, [])
def test_delete_on_edit_se_iri_cannot_delete_non_partial_deposit(self):
"""Delete !partial deposit should return a 400 response
"""
# given
deposit_id = self.create_deposit_ready()
deposit = Deposit.objects.get(pk=deposit_id)
assert deposit.id == deposit_id
# when
url = reverse(EDIT_SE_IRI, args=[self.collection.name, deposit_id])
response = self.client.delete(url)
# then
self.assertEqual(response.status_code,
status.HTTP_400_BAD_REQUEST)
deposit = Deposit.objects.get(pk=deposit_id)
self.assertIsNotNone(deposit)
diff --git a/swh/deposit/tests/api/test_deposit_list.py b/swh/deposit/tests/api/test_deposit_list.py
index 6fb84349..0b21fbdc 100644
--- a/swh/deposit/tests/api/test_deposit_list.py
+++ b/swh/deposit/tests/api/test_deposit_list.py
@@ -1,94 +1,94 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# 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
-from django.core.urlresolvers import reverse
+from django.urls import reverse
import pytest
from rest_framework import status
from rest_framework.test import APITestCase
from swh.deposit.api.converters import convert_status_detail
from ...config import DEPOSIT_STATUS_PARTIAL, PRIVATE_LIST_DEPOSITS
from ..common import BasicTestCase, WithAuthTestCase, CommonCreationRoutine
from ...models import Deposit
@pytest.mark.fs
class CheckDepositListTest(APITestCase, WithAuthTestCase,
BasicTestCase, CommonCreationRoutine):
"""Check deposit list endpoints.
"""
def setUp(self):
super().setUp()
def test_deposit_list(self):
"""Deposit list api should return the deposits
"""
deposit_id = self.create_deposit_partial()
# amend the deposit with a status_detail
deposit = Deposit.objects.get(pk=deposit_id)
status_detail = {
'url': {
'summary': 'At least one compatible url field. Failed',
'fields': ['testurl'],
},
'metadata': [
{
'summary': 'Mandatory fields missing',
'fields': ['9', 10, 1.212],
},
],
'archive': [
{
'summary': 'Invalid archive',
'fields': ['3'],
},
{
'summary': 'Unsupported archive',
'fields': [2],
}
],
}
deposit.status_detail = status_detail
deposit.save()
deposit_id2 = self.create_deposit_partial()
# NOTE: does not work as documented
# https://docs.djangoproject.com/en/1.11/ref/urlresolvers/#django.core.urlresolvers.reverse # noqa
# url = reverse(PRIVATE_LIST_DEPOSITS, kwargs={'page_size': 1})
main_url = reverse(PRIVATE_LIST_DEPOSITS)
url = '%s?page_size=1' % main_url
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data['count'], 2) # 2 deposits
expected_next = '%s?page=2&page_size=1' % main_url
self.assertTrue(data['next'].endswith(expected_next))
self.assertIsNone(data['previous'])
self.assertEqual(len(data['results']), 1) # page of size 1
deposit = data['results'][0]
self.assertEqual(deposit['id'], deposit_id)
self.assertEqual(deposit['status'], DEPOSIT_STATUS_PARTIAL)
expected_status_detail = convert_status_detail(status_detail)
self.assertEqual(deposit['status_detail'], expected_status_detail)
# then 2nd page
response2 = self.client.get(expected_next)
self.assertEqual(response2.status_code, status.HTTP_200_OK)
data2 = response2.json()
self.assertEqual(data2['count'], 2) # still 2 deposits
self.assertIsNone(data2['next'])
expected_previous = '%s?page_size=1' % main_url
self.assertTrue(data2['previous'].endswith(expected_previous))
self.assertEqual(len(data2['results']), 1) # page of size 1
deposit2 = data2['results'][0]
self.assertEqual(deposit2['id'], deposit_id2)
self.assertEqual(deposit2['status'], DEPOSIT_STATUS_PARTIAL)
diff --git a/swh/deposit/tests/api/test_deposit_multipart.py b/swh/deposit/tests/api/test_deposit_multipart.py
index 8502a746..8ba2a2e3 100644
--- a/swh/deposit/tests/api/test_deposit_multipart.py
+++ b/swh/deposit/tests/api/test_deposit_multipart.py
@@ -1,402 +1,402 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# 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
from django.core.files.uploadedfile import InMemoryUploadedFile
-from django.core.urlresolvers import reverse
+from django.urls import reverse
from io import BytesIO
from rest_framework import status
from rest_framework.test import APITestCase
from swh.deposit.config import COL_IRI
from swh.deposit.config import DEPOSIT_STATUS_DEPOSITED
from swh.deposit.models import Deposit, DepositRequest
from swh.deposit.parsers import parse_xml
from ..common import BasicTestCase, WithAuthTestCase
from ..common import FileSystemCreationRoutine
class DepositMultipartTestCase(APITestCase, WithAuthTestCase, BasicTestCase,
FileSystemCreationRoutine):
"""Post multipart deposit scenario
"""
def setUp(self):
super().setUp()
self.data_atom_entry_ok = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom"
xmlns:dcterms="http://purl.org/dc/terms/">
<title>Title</title>
<id>urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a</id>
<updated>2005-10-07T17:17:08Z</updated>
<author><name>Contributor</name></author>
<summary type="text">The abstract</summary>
<!-- some embedded metadata -->
<dcterms:abstract>The abstract</dcterms:abstract>
<dcterms:accessRights>Access Rights</dcterms:accessRights>
<dcterms:alternative>Alternative Title</dcterms:alternative>
<dcterms:available>Date Available</dcterms:available>
<dcterms:bibliographicCitation>Bibliographic Citation</dcterms:bibliographicCitation> # noqa
<dcterms:contributor>Contributor</dcterms:contributor>
<dcterms:description>Description</dcterms:description>
<dcterms:hasPart>Has Part</dcterms:hasPart>
<dcterms:hasVersion>Has Version</dcterms:hasVersion>
<dcterms:identifier>Identifier</dcterms:identifier>
<dcterms:isPartOf>Is Part Of</dcterms:isPartOf>
<dcterms:publisher>Publisher</dcterms:publisher>
<dcterms:references>References</dcterms:references>
<dcterms:rightsHolder>Rights Holder</dcterms:rightsHolder>
<dcterms:source>Source</dcterms:source>
<dcterms:title>Title</dcterms:title>
<dcterms:type>Type</dcterms:type>
</entry>"""
self.data_atom_entry_update_in_place = """<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom"
xmlns:dcterms="http://purl.org/dc/terms/">
<id>urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa7b</id>
<dcterms:title>Title</dcterms:title>
<dcterms:type>Type</dcterms:type>
</entry>"""
def test_post_deposit_multipart_without_slug_header_is_bad_request(self):
# given
url = reverse(COL_IRI, args=[self.collection.name])
data_atom_entry = self.data_atom_entry_ok
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(data_atom_entry),
field_name='atom0',
name='atom0',
content_type='application/atom+xml; charset="utf-8"',
size=len(data_atom_entry),
charset='utf-8')
# when
response = self.client.post(
url,
format='multipart',
data={
'archive': archive,
'atom_entry': atom_entry,
},
# + headers
HTTP_IN_PROGRESS='false')
self.assertIn(b'Missing SLUG header', response.content)
self.assertEqual(response.status_code,
status.HTTP_400_BAD_REQUEST)
def test_post_deposit_multipart_zip(self):
"""one multipart deposit (zip+xml) should be accepted
"""
# given
url = reverse(COL_IRI, args=[self.collection.name])
# from django.core.files import uploadedfile
data_atom_entry = self.data_atom_entry_ok
archive = InMemoryUploadedFile(
BytesIO(self.archive['data']),
field_name=self.archive['name'],
name=self.archive['name'],
content_type='application/zip',
size=self.archive['length'],
charset=None)
atom_entry = InMemoryUploadedFile(
BytesIO(data_atom_entry),
field_name='atom0',
name='atom0',
content_type='application/atom+xml; charset="utf-8"',
size=len(data_atom_entry),
charset='utf-8')
external_id = 'external-id'
# when
response = self.client.post(
url,
format='multipart',
data={
'archive': archive,
'atom_entry': atom_entry,
},
# + headers
HTTP_IN_PROGRESS='false',
HTTP_SLUG=external_id)
# 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_requests = DepositRequest.objects.filter(deposit=deposit)
self.assertEqual(len(deposit_requests), 2)
for deposit_request in deposit_requests:
self.assertEqual(deposit_request.deposit, deposit)
if deposit_request.type == 'archive':
self.assertRegex(deposit_request.archive.name,
self.archive['name'])
self.assertIsNone(deposit_request.metadata)
self.assertIsNone(deposit_request.raw_metadata)
else:
self.assertEqual(
deposit_request.metadata['id'],
'urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a')
self.assertEqual(deposit_request.raw_metadata,
data_atom_entry.decode('utf-8'))
def test_post_deposit_multipart_tar(self):
"""one multipart deposit (tar+xml) should be accepted
"""
# given
url = reverse(COL_IRI, args=[self.collection.name])
# from django.core.files import uploadedfile
data_atom_entry = self.data_atom_entry_ok
archive = InMemoryUploadedFile(
BytesIO(self.archive['data']),
field_name=self.archive['name'],
name=self.archive['name'],
content_type='application/x-tar',
size=self.archive['length'],
charset=None)
atom_entry = InMemoryUploadedFile(
BytesIO(data_atom_entry),
field_name='atom0',
name='atom0',
content_type='application/atom+xml; charset="utf-8"',
size=len(data_atom_entry),
charset='utf-8')
external_id = 'external-id'
# when
response = self.client.post(
url,
format='multipart',
data={
'archive': archive,
'atom_entry': atom_entry,
},
# + headers
HTTP_IN_PROGRESS='false',
HTTP_SLUG=external_id)
# 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_requests = DepositRequest.objects.filter(deposit=deposit)
self.assertEqual(len(deposit_requests), 2)
for deposit_request in deposit_requests:
self.assertEqual(deposit_request.deposit, deposit)
if deposit_request.type == 'archive':
self.assertRegex(deposit_request.archive.name,
self.archive['name'])
self.assertIsNone(deposit_request.metadata)
self.assertIsNone(deposit_request.raw_metadata)
else:
self.assertEqual(
deposit_request.metadata['id'],
'urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a')
self.assertEqual(deposit_request.raw_metadata,
data_atom_entry.decode('utf-8'))
def test_post_deposit_multipart_put_to_replace_metadata(self):
"""One multipart deposit followed by a metadata update should be
accepted
"""
# given
url = reverse(COL_IRI, args=[self.collection.name])
data_atom_entry = self.data_atom_entry_ok
archive = InMemoryUploadedFile(
BytesIO(self.archive['data']),
field_name=self.archive['name'],
name=self.archive['name'],
content_type='application/zip',
size=self.archive['length'],
charset=None)
atom_entry = InMemoryUploadedFile(
BytesIO(data_atom_entry),
field_name='atom0',
name='atom0',
content_type='application/atom+xml; charset="utf-8"',
size=len(data_atom_entry),
charset='utf-8')
external_id = 'external-id'
# when
response = self.client.post(
url,
format='multipart',
data={
'archive': archive,
'atom_entry': atom_entry,
},
# + headers
HTTP_IN_PROGRESS='true',
HTTP_SLUG=external_id)
# 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_requests = DepositRequest.objects.filter(deposit=deposit)
self.assertEqual(len(deposit_requests), 2)
for deposit_request in deposit_requests:
self.assertEqual(deposit_request.deposit, deposit)
if deposit_request.type == 'archive':
self.assertRegex(deposit_request.archive.name,
self.archive['name'])
else:
self.assertEqual(
deposit_request.metadata['id'],
'urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a')
self.assertEqual(deposit_request.raw_metadata,
data_atom_entry.decode('utf-8'))
replace_metadata_uri = response._headers['location'][1]
response = self.client.put(
replace_metadata_uri,
content_type='application/atom+xml;type=entry',
data=self.data_atom_entry_update_in_place,
HTTP_IN_PROGRESS='false')
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
# deposit_id did not change
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 = DepositRequest.objects.filter(deposit=deposit)
self.assertEqual(len(deposit_requests), 2)
for deposit_request in deposit_requests:
self.assertEqual(deposit_request.deposit, deposit)
if deposit_request.type == 'archive':
self.assertRegex(deposit_request.archive.name,
self.archive['name'])
else:
self.assertEqual(
deposit_request.metadata['id'],
'urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa7b')
self.assertEqual(
deposit_request.raw_metadata,
self.data_atom_entry_update_in_place)
# FAILURE scenarios
def test_post_deposit_multipart_only_archive_and_atom_entry(self):
"""Multipart deposit only accepts one archive and one atom+xml"""
# given
url = reverse(COL_IRI, args=[self.collection.name])
archive_content = b'some content representing archive'
archive = InMemoryUploadedFile(BytesIO(archive_content),
field_name='archive0',
name='archive0',
content_type='application/x-tar',
size=len(archive_content),
charset=None)
other_archive_content = b"some-other-content"
other_archive = InMemoryUploadedFile(BytesIO(other_archive_content),
field_name='atom0',
name='atom0',
content_type='application/x-tar',
size=len(other_archive_content),
charset='utf-8')
# when
response = self.client.post(
url,
format='multipart',
data={
'archive': archive,
'atom_entry': other_archive,
},
# + headers
HTTP_IN_PROGRESS='false',
HTTP_SLUG='external-id')
# then
self.assertEqual(response.status_code,
status.HTTP_415_UNSUPPORTED_MEDIA_TYPE)
self.assertTrue(
'Only 1 application/zip (or application/x-tar) archive' in
response.content.decode('utf-8'))
# when
archive.seek(0)
response = self.client.post(
url,
format='multipart',
data={
'archive': archive,
},
# + headers
HTTP_IN_PROGRESS='false',
HTTP_SLUG='external-id')
# then
self.assertEqual(response.status_code,
status.HTTP_415_UNSUPPORTED_MEDIA_TYPE)
self.assertTrue(
'You must provide both 1 application/zip (or '
'application/x-tar) and 1 atom+xml entry for '
'multipart deposit' in response.content.decode('utf-8')
)
diff --git a/swh/deposit/tests/api/test_deposit_read_archive.py b/swh/deposit/tests/api/test_deposit_read_archive.py
index 4c82ad2b..07d16c3f 100644
--- a/swh/deposit/tests/api/test_deposit_read_archive.py
+++ b/swh/deposit/tests/api/test_deposit_read_archive.py
@@ -1,125 +1,125 @@
-# Copyright (C) 2017 The Software Heritage developers
+# 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 hashlib
import os
-from django.core.urlresolvers import reverse
+from django.urls import reverse
import pytest
from rest_framework import status
from rest_framework.test import APITestCase
from swh.core import tarball
from swh.deposit.config import PRIVATE_GET_RAW_CONTENT
from swh.deposit.tests import TEST_CONFIG
from ..common import BasicTestCase, WithAuthTestCase, CommonCreationRoutine
from ..common import FileSystemCreationRoutine, create_arborescence_archive
@pytest.mark.fs
class DepositReadArchivesTest(APITestCase, WithAuthTestCase,
BasicTestCase, CommonCreationRoutine,
FileSystemCreationRoutine):
def setUp(self):
super().setUp()
self.archive2 = create_arborescence_archive(
self.root_path, 'archive2', 'file2', b'some other content in file')
self.workdir = os.path.join(self.root_path, 'workdir')
def test_access_to_existing_deposit_with_one_archive(self):
"""Access to deposit should stream a 200 response with its raw content
"""
deposit_id = self.create_simple_binary_deposit()
url = reverse(PRIVATE_GET_RAW_CONTENT,
args=[self.collection.name, deposit_id])
r = self.client.get(url)
self.assertEqual(r.status_code, status.HTTP_200_OK)
self.assertEqual(r._headers['content-type'][1],
'application/octet-stream')
# read the stream
data = b''.join(r.streaming_content)
actual_sha1 = hashlib.sha1(data).hexdigest()
self.assertEqual(actual_sha1, self.archive['sha1sum'])
# this does not touch the extraction dir so this should stay empty
self.assertEqual(os.listdir(TEST_CONFIG['extraction_dir']), [])
def _check_tarball_consistency(self, actual_sha1):
tarball.uncompress(self.archive['path'], self.workdir)
self.assertEqual(os.listdir(self.workdir), ['file1'])
tarball.uncompress(self.archive2['path'], self.workdir)
lst = set(os.listdir(self.workdir))
self.assertEqual(lst, {'file1', 'file2'})
new_path = self.workdir + '.zip'
tarball.compress(new_path, 'zip', self.workdir)
with open(new_path, 'rb') as f:
h = hashlib.sha1(f.read()).hexdigest()
self.assertEqual(actual_sha1, h)
self.assertNotEqual(actual_sha1, self.archive['sha1sum'])
self.assertNotEqual(actual_sha1, self.archive2['sha1sum'])
def test_access_to_existing_deposit_with_multiple_archives(self):
"""Access to deposit should stream a 200 response with its raw contents
"""
deposit_id = self.create_complex_binary_deposit()
url = reverse(PRIVATE_GET_RAW_CONTENT,
args=[self.collection.name, deposit_id])
r = self.client.get(url)
self.assertEqual(r.status_code, status.HTTP_200_OK)
self.assertEqual(r._headers['content-type'][1],
'application/octet-stream')
# read the stream
data = b''.join(r.streaming_content)
actual_sha1 = hashlib.sha1(data).hexdigest()
self._check_tarball_consistency(actual_sha1)
# this touches the extraction directory but should clean up
# after itself
self.assertEqual(os.listdir(TEST_CONFIG['extraction_dir']), [])
class DepositReadArchivesFailureTest(APITestCase, WithAuthTestCase,
BasicTestCase, CommonCreationRoutine):
def test_access_to_nonexisting_deposit_returns_404_response(self):
"""Read unknown collection should return a 404 response
"""
unknown_id = '999'
url = reverse(PRIVATE_GET_RAW_CONTENT,
args=[self.collection.name, unknown_id])
response = self.client.get(url)
self.assertEqual(response.status_code,
status.HTTP_404_NOT_FOUND)
self.assertIn('Deposit with id %s does not exist' % unknown_id,
response.content.decode('utf-8'))
def test_access_to_nonexisting_collection_returns_404_response(self):
"""Read unknown deposit should return a 404 response
"""
collection_name = 'non-existing'
deposit_id = self.create_deposit_partial()
url = reverse(PRIVATE_GET_RAW_CONTENT,
args=[collection_name, deposit_id])
response = self.client.get(url)
self.assertEqual(response.status_code,
status.HTTP_404_NOT_FOUND)
self.assertIn('Unknown collection name %s' % collection_name,
response.content.decode('utf-8'))
diff --git a/swh/deposit/tests/api/test_deposit_read_metadata.py b/swh/deposit/tests/api/test_deposit_read_metadata.py
index 9f40e701..e35200bd 100644
--- a/swh/deposit/tests/api/test_deposit_read_metadata.py
+++ b/swh/deposit/tests/api/test_deposit_read_metadata.py
@@ -1,205 +1,205 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# 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
-from django.core.urlresolvers import reverse
+from django.urls import reverse
from rest_framework import status
from rest_framework.test import APITestCase
from swh.deposit.models import Deposit
from swh.deposit.config import PRIVATE_GET_DEPOSIT_METADATA
from swh.deposit.config import DEPOSIT_STATUS_LOAD_SUCCESS
from swh.deposit.config import DEPOSIT_STATUS_PARTIAL
from ...config import SWH_PERSON
from ..common import BasicTestCase, WithAuthTestCase, CommonCreationRoutine
class DepositReadMetadataTest(APITestCase, WithAuthTestCase, BasicTestCase,
CommonCreationRoutine):
"""Deposit access to read metadata information on deposit.
"""
def test_read_metadata(self):
"""Private metadata read api to existing deposit should return metadata
"""
deposit_id = self.create_deposit_partial()
url = reverse(PRIVATE_GET_DEPOSIT_METADATA,
args=[self.collection.name, deposit_id])
response = self.client.get(url)
self.assertEqual(response.status_code,
status.HTTP_200_OK)
self.assertEqual(response._headers['content-type'][1],
'application/json')
data = response.json()
expected_meta = {
'origin': {
'url': 'https://hal-test.archives-ouvertes.fr/' +
'some-external-id',
'type': 'deposit'
},
'origin_metadata': {
'metadata': {
'@xmlns': ['http://www.w3.org/2005/Atom'],
'author': ['some awesome author', 'another one', 'no one'],
'external_identifier': 'some-external-id',
'url': 'https://hal-test.archives-ouvertes.fr/' +
'some-external-id'
},
'provider': {
'provider_name': 'hal',
'provider_type': 'deposit_client',
'provider_url': 'https://hal-test.archives-ouvertes.fr/',
'metadata': {}
},
'tool': {
'name': 'swh-deposit',
'version': '0.0.1',
'configuration': {
'sword_version': '2'
}
}
},
'revision': {
'synthetic': True,
'committer_date': None,
'message': 'hal: Deposit %s in collection hal' % deposit_id,
'author': SWH_PERSON,
'committer': SWH_PERSON,
'date': None,
'metadata': {
'@xmlns': ['http://www.w3.org/2005/Atom'],
'author': ['some awesome author', 'another one', 'no one'],
'external_identifier': 'some-external-id',
'url': 'https://hal-test.archives-ouvertes.fr/' +
'some-external-id'
},
'type': 'tar'
},
'branch_name': 'master',
}
self.assertEqual(data, expected_meta)
def test_read_metadata_revision_with_parent(self):
"""Private read metadata to a deposit (with parent) returns metadata
"""
swh_id = 'da78a9d4cf1d5d29873693fd496142e3a18c20fa'
swh_persistent_id = 'swh:1:rev:%s' % swh_id
deposit_id1 = self.create_deposit_with_status(
status=DEPOSIT_STATUS_LOAD_SUCCESS,
external_id='some-external-id',
swh_id=swh_persistent_id)
deposit_parent = Deposit.objects.get(pk=deposit_id1)
self.assertEqual(deposit_parent.swh_id, swh_persistent_id)
self.assertEqual(deposit_parent.external_id, 'some-external-id')
self.assertEqual(deposit_parent.status, DEPOSIT_STATUS_LOAD_SUCCESS)
deposit_id = self.create_deposit_partial(
external_id='some-external-id')
deposit = Deposit.objects.get(pk=deposit_id)
self.assertEqual(deposit.external_id, 'some-external-id')
self.assertEqual(deposit.swh_id, None)
self.assertEqual(deposit.parent, deposit_parent)
self.assertEqual(deposit.status, DEPOSIT_STATUS_PARTIAL)
url = reverse(PRIVATE_GET_DEPOSIT_METADATA,
args=[self.collection.name, deposit_id])
response = self.client.get(url)
self.assertEqual(response.status_code,
status.HTTP_200_OK)
self.assertEqual(response._headers['content-type'][1],
'application/json')
data = response.json()
expected_meta = {
'origin': {
'url': 'https://hal-test.archives-ouvertes.fr/' +
'some-external-id',
'type': 'deposit'
},
'origin_metadata': {
'metadata': {
'@xmlns': ['http://www.w3.org/2005/Atom'],
'author': ['some awesome author', 'another one', 'no one'],
'external_identifier': 'some-external-id',
'url': 'https://hal-test.archives-ouvertes.fr/' +
'some-external-id'
},
'provider': {
'provider_name': 'hal',
'provider_type': 'deposit_client',
'provider_url': 'https://hal-test.archives-ouvertes.fr/',
'metadata': {}
},
'tool': {
'name': 'swh-deposit',
'version': '0.0.1',
'configuration': {
'sword_version': '2'
}
}
},
'revision': {
'synthetic': True,
'date': None,
'committer_date': None,
'author': SWH_PERSON,
'committer': SWH_PERSON,
'type': 'tar',
'message': 'hal: Deposit %s in collection hal' % deposit_id,
'metadata': {
'@xmlns': ['http://www.w3.org/2005/Atom'],
'author': ['some awesome author', 'another one', 'no one'],
'external_identifier': 'some-external-id',
'url': 'https://hal-test.archives-ouvertes.fr/' +
'some-external-id'
},
'parents': [swh_id]
},
'branch_name': 'master',
}
self.assertEqual(data, expected_meta)
def test_access_to_nonexisting_deposit_returns_404_response(self):
"""Read unknown collection should return a 404 response
"""
unknown_id = '999'
url = reverse(PRIVATE_GET_DEPOSIT_METADATA,
args=[self.collection.name, unknown_id])
response = self.client.get(url)
self.assertEqual(response.status_code,
status.HTTP_404_NOT_FOUND)
self.assertIn('Deposit with id %s does not exist' % unknown_id,
response.content.decode('utf-8'))
def test_access_to_nonexisting_collection_returns_404_response(self):
"""Read unknown deposit should return a 404 response
"""
collection_name = 'non-existing'
deposit_id = self.create_deposit_partial()
url = reverse(PRIVATE_GET_DEPOSIT_METADATA,
args=[collection_name, deposit_id])
response = self.client.get(url)
self.assertEqual(response.status_code,
status.HTTP_404_NOT_FOUND)
self.assertIn('Unknown collection name %s' % collection_name,
response.content.decode('utf-8'),)
diff --git a/swh/deposit/tests/api/test_deposit_status.py b/swh/deposit/tests/api/test_deposit_status.py
index 256f2cf6..f6fa3d1e 100644
--- a/swh/deposit/tests/api/test_deposit_status.py
+++ b/swh/deposit/tests/api/test_deposit_status.py
@@ -1,144 +1,144 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# 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
-from django.core.urlresolvers import reverse
+from django.urls import reverse
from io import BytesIO
from rest_framework import status
from rest_framework.test import APITestCase
from swh.deposit.config import (COL_IRI, STATE_IRI, DEPOSIT_STATUS_DEPOSITED,
DEPOSIT_STATUS_REJECTED)
from swh.deposit.models import Deposit, DEPOSIT_STATUS_DETAIL
from swh.deposit.models import DEPOSIT_STATUS_LOAD_SUCCESS
from swh.deposit.parsers import parse_xml
from ..common import BasicTestCase, WithAuthTestCase, FileSystemCreationRoutine
from ..common import CommonCreationRoutine
class DepositStatusTestCase(APITestCase, WithAuthTestCase, BasicTestCase,
FileSystemCreationRoutine, CommonCreationRoutine):
"""Status on deposit
"""
def test_post_deposit_with_status_check(self):
"""Binary upload should be accepted
"""
# 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='false',
HTTP_CONTENT_DISPOSITION='attachment; filename=filename0')
# then
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
deposit = Deposit.objects.get(external_id=external_id)
status_url = reverse(STATE_IRI,
args=[self.collection.name, deposit.id])
# check status
status_response = self.client.get(status_url)
self.assertEqual(status_response.status_code, status.HTTP_200_OK)
r = parse_xml(BytesIO(status_response.content))
self.assertEqual(int(r['deposit_id']), deposit.id)
self.assertEqual(r['deposit_status'], DEPOSIT_STATUS_DEPOSITED)
self.assertEqual(r['deposit_status_detail'],
DEPOSIT_STATUS_DETAIL[DEPOSIT_STATUS_DEPOSITED])
def test_status_with_swh_information(self):
_status = DEPOSIT_STATUS_LOAD_SUCCESS
_context = 'https://hal.archives-ouvertes.fr/hal-01727745'
_swh_id = 'swh:1:dir:42a13fc721c8716ff695d0d62fc851d641f3a12b'
_swh_id_context = '%s;%s' % (_swh_id, _context)
_swh_anchor_id = 'swh:rev:1:548b3c0a2bb43e1fca191e24b5803ff6b3bc7c10'
_swh_anchor_id_context = '%s;%s' % (_swh_anchor_id, _context)
# given
deposit_id = self.create_deposit_with_status(
status=_status,
swh_id=_swh_id,
swh_id_context=_swh_id_context,
swh_anchor_id=_swh_anchor_id,
swh_anchor_id_context=_swh_anchor_id_context
)
url = reverse(STATE_IRI, args=[self.collection.name, deposit_id])
# when
status_response = self.client.get(url)
# then
self.assertEqual(status_response.status_code, status.HTTP_200_OK)
r = parse_xml(BytesIO(status_response.content))
self.assertEqual(int(r['deposit_id']), deposit_id)
self.assertEqual(r['deposit_status'], _status)
self.assertEqual(r['deposit_status_detail'],
DEPOSIT_STATUS_DETAIL[DEPOSIT_STATUS_LOAD_SUCCESS])
self.assertEqual(r['deposit_swh_id'], _swh_id)
self.assertEqual(r['deposit_swh_id_context'], _swh_id_context)
self.assertEqual(r['deposit_swh_anchor_id'], _swh_anchor_id)
self.assertEqual(r['deposit_swh_anchor_id_context'],
_swh_anchor_id_context)
def test_status_on_unknown_deposit(self):
"""Asking for the status of unknown deposit returns 404 response"""
status_url = reverse(STATE_IRI, args=[self.collection.name, 999])
status_response = self.client.get(status_url)
self.assertEqual(status_response.status_code,
status.HTTP_404_NOT_FOUND)
def test_status_with_http_accept_header_should_not_break(self):
"""Asking deposit status with Accept header should return 200
"""
deposit_id = self.create_deposit_partial()
status_url = reverse(STATE_IRI, args=[
self.collection.name, deposit_id])
response = self.client.get(
status_url,
HTTP_ACCEPT='text/html,application/xml;q=9,*/*,q=8')
self.assertEqual(response.status_code, status.HTTP_200_OK)
def test_status_on_deposit_rejected(self):
_status = DEPOSIT_STATUS_REJECTED
_swh_id = '548b3c0a2bb43e1fca191e24b5803ff6b3bc7c10'
_status_detail = {'url': {'summary': 'Wrong url'}}
# given
deposit_id = self.create_deposit_with_status(
status=_status, swh_id=_swh_id, status_detail=_status_detail)
url = reverse(STATE_IRI, args=[self.collection.name, deposit_id])
# when
status_response = self.client.get(url)
# then
self.assertEqual(status_response.status_code, status.HTTP_200_OK)
r = parse_xml(BytesIO(status_response.content))
self.assertEqual(int(r['deposit_id']), deposit_id)
self.assertEqual(r['deposit_status'], _status)
self.assertEqual(r['deposit_status_detail'], '- Wrong url')
self.assertEqual(r['deposit_swh_id'], _swh_id)
diff --git a/swh/deposit/tests/api/test_deposit_update.py b/swh/deposit/tests/api/test_deposit_update.py
index 341936a6..227c1a2d 100644
--- a/swh/deposit/tests/api/test_deposit_update.py
+++ b/swh/deposit/tests/api/test_deposit_update.py
@@ -1,333 +1,333 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# 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
-from django.core.urlresolvers import reverse
+from django.urls import reverse
from rest_framework import status
from rest_framework.test import APITestCase
from swh.deposit.models import Deposit, DepositRequest
from swh.deposit.config import EDIT_SE_IRI, EM_IRI
from ..common import BasicTestCase, WithAuthTestCase, CommonCreationRoutine
from ..common import FileSystemCreationRoutine, create_arborescence_archive
class DepositUpdateOrReplaceExistingDataTest(
APITestCase, WithAuthTestCase, BasicTestCase,
FileSystemCreationRoutine, CommonCreationRoutine):
"""Try put/post (update/replace) query on EM_IRI
"""
def setUp(self):
super().setUp()
self.atom_entry_data1 = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom">
<foobar>bar</foobar>
</entry>"""
self.atom_entry_data1 = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom">
<foobar>bar</foobar>
</entry>"""
self.archive2 = create_arborescence_archive(
self.root_path, 'archive2', 'file2', b'some other content in file')
def test_replace_archive_to_deposit_is_possible(self):
"""Replace all archive with another one should return a 204 response
"""
# given
deposit_id = self.create_simple_binary_deposit(status_partial=True)
deposit = Deposit.objects.get(pk=deposit_id)
requests = DepositRequest.objects.filter(
deposit=deposit,
type='archive')
assert len(list(requests)) == 1
assert self.archive['name'] in requests[0].archive.name
# we have no metadata for that deposit
requests = list(DepositRequest.objects.filter(
deposit=deposit, type='metadata'))
assert len(requests) == 0
deposit_id = self._update_deposit_with_status(deposit_id,
status_partial=True)
requests = list(DepositRequest.objects.filter(
deposit=deposit, type='metadata'))
assert len(requests) == 1
update_uri = reverse(EM_IRI, args=[self.collection.name, deposit_id])
external_id = 'some-external-id-1'
response = self.client.put(
update_uri,
content_type='application/zip', # as zip
data=self.archive2['data'],
# + headers
CONTENT_LENGTH=self.archive2['length'],
HTTP_SLUG=external_id,
HTTP_CONTENT_MD5=self.archive2['md5sum'],
HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip',
HTTP_IN_PROGRESS='false',
HTTP_CONTENT_DISPOSITION='attachment; filename=%s' % (
self.archive2['name'], ))
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
requests = DepositRequest.objects.filter(
deposit=deposit,
type='archive')
self.assertEqual(len(list(requests)), 1)
self.assertRegex(requests[0].archive.name, self.archive2['name'])
# check we did not touch the other parts
requests = list(DepositRequest.objects.filter(
deposit=deposit, type='metadata'))
self.assertEqual(len(requests), 1)
def test_replace_metadata_to_deposit_is_possible(self):
"""Replace all metadata with another one should return a 204 response
"""
# given
deposit_id = self.create_simple_binary_deposit(status_partial=True)
deposit = Deposit.objects.get(pk=deposit_id)
requests = DepositRequest.objects.filter(
deposit=deposit,
type='metadata')
assert len(list(requests)) == 0
requests = list(DepositRequest.objects.filter(
deposit=deposit, type='archive'))
assert len(requests) == 1
update_uri = reverse(EDIT_SE_IRI, args=[self.collection.name,
deposit_id])
response = self.client.put(
update_uri,
content_type='application/atom+xml;type=entry',
data=self.atom_entry_data1)
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
requests = DepositRequest.objects.filter(
deposit=deposit,
type='metadata')
self.assertEqual(len(list(requests)), 1)
metadata = requests[0].metadata
self.assertEqual(metadata['foobar'], 'bar')
# check we did not touch the other parts
requests = list(DepositRequest.objects.filter(
deposit=deposit, type='archive'))
self.assertEqual(len(requests), 1)
def test_add_archive_to_deposit_is_possible(self):
"""Add another archive to a deposit return a 201 response
"""
# given
deposit_id = self.create_simple_binary_deposit(status_partial=True)
deposit = Deposit.objects.get(pk=deposit_id)
requests = DepositRequest.objects.filter(
deposit=deposit,
type='archive')
assert len(list(requests)) == 1
assert self.archive['name'] in requests[0].archive.name
requests = list(DepositRequest.objects.filter(
deposit=deposit, type='metadata'))
assert len(requests) == 0
update_uri = reverse(EM_IRI, args=[self.collection.name, deposit_id])
external_id = 'some-external-id-1'
response = self.client.post(
update_uri,
content_type='application/zip', # as zip
data=self.archive2['data'],
# + headers
CONTENT_LENGTH=self.archive2['length'],
HTTP_SLUG=external_id,
HTTP_CONTENT_MD5=self.archive2['md5sum'],
HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip',
HTTP_IN_PROGRESS='false',
HTTP_CONTENT_DISPOSITION='attachment; filename=%s' % (
self.archive2['name'],))
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
requests = list(DepositRequest.objects.filter(
deposit=deposit,
type='archive').order_by('id'))
self.assertEqual(len(requests), 2)
# first archive still exists
self.assertRegex(requests[0].archive.name, self.archive['name'])
# a new one was added
self.assertRegex(requests[1].archive.name, self.archive2['name'])
# check we did not touch the other parts
requests = list(DepositRequest.objects.filter(
deposit=deposit, type='metadata'))
self.assertEqual(len(requests), 0)
def test_add_metadata_to_deposit_is_possible(self):
"""Add metadata with another one should return a 204 response
"""
# given
deposit_id = self.create_deposit_partial()
deposit = Deposit.objects.get(pk=deposit_id)
requests = DepositRequest.objects.filter(
deposit=deposit,
type='metadata')
assert len(list(requests)) == 2
requests = list(DepositRequest.objects.filter(
deposit=deposit, type='archive'))
assert len(requests) == 0
update_uri = reverse(EDIT_SE_IRI, args=[self.collection.name,
deposit_id])
response = self.client.post(
update_uri,
content_type='application/atom+xml;type=entry',
data=self.atom_entry_data1)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
requests = DepositRequest.objects.filter(
deposit=deposit,
type='metadata').order_by('id')
self.assertEqual(len(list(requests)), 3)
# a new one was added
self.assertEqual(requests[1].metadata['foobar'], 'bar')
# check we did not touch the other parts
requests = list(DepositRequest.objects.filter(
deposit=deposit, type='archive'))
self.assertEqual(len(requests), 0)
class DepositUpdateFailuresTest(APITestCase, WithAuthTestCase, BasicTestCase,
CommonCreationRoutine):
"""Failure scenario about add/replace (post/put) query on deposit.
"""
def test_add_metadata_to_unknown_collection(self):
"""Replacing metadata to unknown deposit should return a 404 response
"""
url = reverse(EDIT_SE_IRI, args=['test', 1000])
response = self.client.post(
url,
content_type='application/atom+xml;type=entry',
data=self.atom_entry_data0)
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self.assertRegex(response.content.decode('utf-8'),
'Unknown collection name test')
def test_add_metadata_to_unknown_deposit(self):
"""Replacing metadata to unknown deposit should return a 404 response
"""
url = reverse(EDIT_SE_IRI, args=[self.collection.name, 999])
response = self.client.post(
url,
content_type='application/atom+xml;type=entry',
data=self.atom_entry_data0)
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self.assertRegex(response.content.decode('utf-8'),
'Deposit with id 999 does not exist')
def test_replace_metadata_to_unknown_deposit(self):
"""Adding metadata to unknown deposit should return a 404 response
"""
url = reverse(EDIT_SE_IRI, args=[self.collection.name, 998])
response = self.client.put(
url,
content_type='application/atom+xml;type=entry',
data=self.atom_entry_data0)
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self.assertRegex(response.content.decode('utf-8'),
'Deposit with id 998 does not exist')
def test_add_archive_to_unknown_deposit(self):
"""Adding metadata to unknown deposit should return a 404 response
"""
url = reverse(EM_IRI, args=[self.collection.name, 997])
response = self.client.post(
url,
content_type='application/zip',
data=self.atom_entry_data0)
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self.assertRegex(response.content.decode('utf-8'),
'Deposit with id 997 does not exist')
def test_replace_archive_to_unknown_deposit(self):
"""Replacing archive to unknown deposit should return a 404 response
"""
url = reverse(EM_IRI, args=[self.collection.name, 996])
response = self.client.put(
url,
content_type='application/zip',
data=self.atom_entry_data0)
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self.assertRegex(response.content.decode('utf-8'),
'Deposit with id 996 does not exist')
def test_post_metadata_to_em_iri_failure(self):
"""Update (POST) archive with wrong content type should return 400
"""
deposit_id = self.create_deposit_partial() # only update on partial
update_uri = reverse(EM_IRI, args=[self.collection.name, deposit_id])
response = self.client.post(
update_uri,
content_type='application/x-gtar-compressed',
data=self.atom_entry_data0)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertRegex(response.content.decode('utf-8'),
'Packaging format supported is restricted to '
'application/zip, application/x-tar')
def test_put_metadata_to_em_iri_failure(self):
"""Update (PUT) archive with wrong content type should return 400
"""
# given
deposit_id = self.create_deposit_partial() # only update on partial
# when
update_uri = reverse(EM_IRI, args=[self.collection.name, deposit_id])
response = self.client.put(
update_uri,
content_type='application/atom+xml;type=entry',
data=self.atom_entry_data0)
# then
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertRegex(response.content.decode('utf-8'),
'Packaging format supported is restricted to '
'application/zip, application/x-tar')
diff --git a/swh/deposit/tests/api/test_deposit_update_status.py b/swh/deposit/tests/api/test_deposit_update_status.py
index 1ec4dbdb..d338ceab 100644
--- a/swh/deposit/tests/api/test_deposit_update_status.py
+++ b/swh/deposit/tests/api/test_deposit_update_status.py
@@ -1,130 +1,130 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# 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 json
-from django.core.urlresolvers import reverse
+from django.urls import reverse
from rest_framework import status
from rest_framework.test import APITestCase
from swh.deposit.models import Deposit, DEPOSIT_STATUS_DETAIL
from swh.deposit.config import PRIVATE_PUT_DEPOSIT, DEPOSIT_STATUS_VERIFIED
from swh.deposit.config import DEPOSIT_STATUS_LOAD_SUCCESS
from ..common import BasicTestCase
class UpdateDepositStatusTest(APITestCase, BasicTestCase):
"""Update the deposit's status scenario
"""
def setUp(self):
super().setUp()
deposit = Deposit(status=DEPOSIT_STATUS_VERIFIED,
collection=self.collection,
client=self.user)
deposit.save()
self.deposit = Deposit.objects.get(pk=deposit.id)
assert self.deposit.status == DEPOSIT_STATUS_VERIFIED
def test_update_deposit_status(self):
"""Existing status for update should return a 204 response
"""
url = reverse(PRIVATE_PUT_DEPOSIT,
args=[self.collection.name, self.deposit.id])
possible_status = set(DEPOSIT_STATUS_DETAIL.keys()) - set(
[DEPOSIT_STATUS_LOAD_SUCCESS])
for _status in possible_status:
response = self.client.put(
url,
content_type='application/json',
data=json.dumps({'status': _status}))
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
deposit = Deposit.objects.get(pk=self.deposit.id)
self.assertEqual(deposit.status, _status)
def test_update_deposit_status_with_info(self):
"""Existing status for update with info should return a 204 response
"""
url = reverse(PRIVATE_PUT_DEPOSIT,
args=[self.collection.name, self.deposit.id])
expected_status = DEPOSIT_STATUS_LOAD_SUCCESS
origin_url = 'something'
directory_id = '42a13fc721c8716ff695d0d62fc851d641f3a12b'
revision_id = '47dc6b4636c7f6cba0df83e3d5490bf4334d987e'
expected_swh_id = 'swh:1:dir:%s' % directory_id
expected_swh_id_context = 'swh:1:dir:%s;origin=%s' % (
directory_id, origin_url)
expected_swh_anchor_id = 'swh:1:rev:%s' % revision_id
expected_swh_anchor_id_context = 'swh:1:rev:%s;origin=%s' % (
revision_id, origin_url)
response = self.client.put(
url,
content_type='application/json',
data=json.dumps({
'status': expected_status,
'revision_id': revision_id,
'directory_id': directory_id,
'origin_url': origin_url,
}))
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
deposit = Deposit.objects.get(pk=self.deposit.id)
self.assertEqual(deposit.status, expected_status)
self.assertEqual(deposit.swh_id, expected_swh_id)
self.assertEqual(deposit.swh_id_context, expected_swh_id_context)
self.assertEqual(deposit.swh_anchor_id, expected_swh_anchor_id)
self.assertEqual(deposit.swh_anchor_id_context,
expected_swh_anchor_id_context)
def test_update_deposit_status_will_fail_with_unknown_status(self):
"""Unknown status for update should return a 400 response
"""
url = reverse(PRIVATE_PUT_DEPOSIT,
args=[self.collection.name, self.deposit.id])
response = self.client.put(
url,
content_type='application/json',
data=json.dumps({'status': 'unknown'}))
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
def test_update_deposit_status_will_fail_with_no_status_key(self):
"""No status provided for update should return a 400 response
"""
url = reverse(PRIVATE_PUT_DEPOSIT,
args=[self.collection.name, self.deposit.id])
response = self.client.put(
url,
content_type='application/json',
data=json.dumps({'something': 'something'}))
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
def test_update_deposit_status_success_without_swh_id_fail(self):
"""Providing successful status without swh_id should return a 400
"""
url = reverse(PRIVATE_PUT_DEPOSIT,
args=[self.collection.name, self.deposit.id])
response = self.client.put(
url,
content_type='application/json',
data=json.dumps({'status': DEPOSIT_STATUS_LOAD_SUCCESS}))
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
diff --git a/swh/deposit/tests/api/test_service_document.py b/swh/deposit/tests/api/test_service_document.py
index 3afa1ca5..61d8e074 100644
--- a/swh/deposit/tests/api/test_service_document.py
+++ b/swh/deposit/tests/api/test_service_document.py
@@ -1,102 +1,102 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# 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
-from django.core.urlresolvers import reverse
+from django.urls import reverse
from rest_framework import status
from rest_framework.test import APITestCase
from swh.deposit.tests import TEST_CONFIG
from swh.deposit.config import SD_IRI
from ..common import BasicTestCase, WithAuthTestCase
class ServiceDocumentNoAuthCase(APITestCase, BasicTestCase):
"""Service document endpoints are protected with basic authentication.
"""
def test_service_document_no_authentication_fails(self):
"""Without authentication, service document endpoint should return 401
"""
url = reverse(SD_IRI)
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
def test_service_document_with_http_accept_should_not_break(self):
"""Without auth, sd endpoint through browser should return 401
"""
url = reverse(SD_IRI)
# when
response = self.client.get(
url,
HTTP_ACCEPT='text/html,application/xml;q=9,*/*,q=8')
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
class ServiceDocumentCase(APITestCase, WithAuthTestCase, BasicTestCase):
def assertResponseOk(self, response): # noqa: N802
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.content.decode('utf-8'),
'''<?xml version="1.0" ?>
<service xmlns:dcterms="http://purl.org/dc/terms/"
xmlns:sword="http://purl.org/net/sword/terms/"
xmlns:atom="http://www.w3.org/2005/Atom"
xmlns="http://www.w3.org/2007/app">
<sword:version>2.0</sword:version>
<sword:maxUploadSize>%s</sword:maxUploadSize>
<workspace>
<atom:title>The Software Heritage (SWH) Archive</atom:title>
<collection href="http://testserver/1/%s/">
<atom:title>%s Software Collection</atom:title>
<accept>application/zip</accept>
<accept>application/x-tar</accept>
<sword:collectionPolicy>Collection Policy</sword:collectionPolicy>
<dcterms:abstract>Software Heritage Archive</dcterms:abstract>
<sword:treatment>Collect, Preserve, Share</sword:treatment>
<sword:mediation>false</sword:mediation>
<sword:metadataRelevantHeader>false</sword:metadataRelevantHeader>
<sword:acceptPackaging>http://purl.org/net/sword/package/SimpleZip</sword:acceptPackaging>
<sword:service>http://testserver/1/%s/</sword:service>
<sword:name>%s</sword:name>
</collection>
</workspace>
</service>
''' % (TEST_CONFIG['max_upload_size'],
self.username,
self.username,
self.username,
self.username)) # noqa
def test_service_document(self):
"""With authentication, service document list user's collection
"""
url = reverse(SD_IRI)
# when
response = self.client.get(url)
# then
self.assertResponseOk(response)
def test_service_document_with_http_accept_header(self):
"""With authentication, with browser, sd list user's collection
"""
url = reverse(SD_IRI)
# when
response = self.client.get(
url,
HTTP_ACCEPT='text/html,application/xml;q=9,*/*,q=8')
self.assertResponseOk(response)
diff --git a/swh/deposit/tests/common.py b/swh/deposit/tests/common.py
index 8a847e63..5c8b70fa 100644
--- a/swh/deposit/tests/common.py
+++ b/swh/deposit/tests/common.py
@@ -1,564 +1,564 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# 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 base64
import hashlib
import os
import shutil
import tarfile
import tempfile
-from django.core.urlresolvers import reverse
+from django.urls import reverse
from django.test import TestCase
from io import BytesIO
import pytest
from rest_framework import status
from swh.deposit.config import (COL_IRI, EM_IRI, EDIT_SE_IRI,
DEPOSIT_STATUS_PARTIAL,
DEPOSIT_STATUS_VERIFIED,
DEPOSIT_STATUS_REJECTED,
DEPOSIT_STATUS_DEPOSITED)
from swh.deposit.models import DepositClient, DepositCollection, Deposit
from swh.deposit.models import DepositRequest
from swh.deposit.parsers import parse_xml
from swh.deposit.settings.testing import MEDIA_ROOT
from swh.core import tarball
def compute_info(archive_path):
"""Given a path, compute information on path.
"""
with open(archive_path, 'rb') as f:
length = 0
sha1sum = hashlib.sha1()
md5sum = hashlib.md5()
data = b''
for chunk in f:
sha1sum.update(chunk)
md5sum.update(chunk)
length += len(chunk)
data += chunk
return {
'dir': os.path.dirname(archive_path),
'name': os.path.basename(archive_path),
'path': archive_path,
'length': length,
'sha1sum': sha1sum.hexdigest(),
'md5sum': md5sum.hexdigest(),
'data': data
}
def _compress(path, extension, dir_path):
"""Compress path according to extension
"""
if extension == 'zip' or extension == 'tar':
return tarball.compress(path, extension, dir_path)
elif '.' in extension:
split_ext = extension.split('.')
if split_ext[0] != 'tar':
raise ValueError(
'Development error, only zip or tar archive supported, '
'%s not supported' % extension)
# deal with specific tar
mode = split_ext[1]
supported_mode = ['xz', 'gz', 'bz2']
if mode not in supported_mode:
raise ValueError(
'Development error, only %s supported, %s not supported' % (
supported_mode, mode))
files = tarball._ls(dir_path)
with tarfile.open(path, 'w:%s' % mode) as t:
for fpath, fname in files:
t.add(fpath, arcname=fname, recursive=False)
return path
def create_arborescence_archive(root_path, archive_name, filename, content,
up_to_size=None, extension='zip'):
"""Build an archive named archive_name in the root_path.
This archive contains one file named filename with the content content.
Args:
root_path (str): Location path of the archive to create
archive_name (str): Archive's name (without extension)
filename (str): Archive's content is only one filename
content (bytes): Content of the filename
up_to_size (int | None): Fill in the blanks size to oversize
or complete an archive's size
extension (str): Extension of the archive to write (default is zip)
Returns:
dict with the keys:
- dir: the directory of that archive
- path: full path to the archive
- sha1sum: archive's sha1sum
- length: archive's length
"""
os.makedirs(root_path, exist_ok=True)
archive_path_dir = tempfile.mkdtemp(dir=root_path)
dir_path = os.path.join(archive_path_dir, archive_name)
os.mkdir(dir_path)
filepath = os.path.join(dir_path, filename)
_length = len(content)
count = 0
batch_size = 128
with open(filepath, 'wb') as f:
f.write(content)
if up_to_size: # fill with blank content up to a given size
count += _length
while count < up_to_size:
f.write(b'0'*batch_size)
count += batch_size
_path = '%s.%s' % (dir_path, extension)
_path = _compress(_path, extension, dir_path)
return compute_info(_path)
def create_archive_with_archive(root_path, name, archive):
"""Create an archive holding another.
"""
invalid_archive_path = os.path.join(root_path, name)
with tarfile.open(invalid_archive_path, 'w:gz') as _archive:
_archive.add(archive['path'], arcname=archive['name'])
return compute_info(invalid_archive_path)
@pytest.mark.fs
class FileSystemCreationRoutine(TestCase):
"""Mixin intended for tests needed to tamper with archives.
"""
def setUp(self):
"""Define the test client and other test variables."""
super().setUp()
self.root_path = '/tmp/swh-deposit/test/build-zip/'
os.makedirs(self.root_path, exist_ok=True)
self.archive = create_arborescence_archive(
self.root_path, 'archive1', 'file1', b'some content in file')
self.atom_entry = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom">
<title>Awesome Compiler</title>
<id>urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a</id>
<external_identifier>1785io25c695</external_identifier>
<updated>2017-10-07T15:17:08Z</updated>
<author>some awesome author</author>
<url>https://hal-test.archives-ouvertes.fr</url>
</entry>"""
def tearDown(self):
super().tearDown()
shutil.rmtree(self.root_path)
def create_simple_binary_deposit(self, status_partial=True):
response = self.client.post(
reverse(COL_IRI, args=[self.collection.name]),
content_type='application/zip',
data=self.archive['data'],
CONTENT_LENGTH=self.archive['length'],
HTTP_MD5SUM=self.archive['md5sum'],
HTTP_SLUG='external-id',
HTTP_IN_PROGRESS=status_partial,
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))
_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'])
return deposit_id
def create_complex_binary_deposit(self, status_partial=False):
deposit_id = self.create_simple_binary_deposit(
status_partial=True)
# Add a second archive to the deposit
# update its status to DEPOSIT_STATUS_VERIFIED
response = self.client.post(
reverse(EM_IRI, args=[self.collection.name, deposit_id]),
content_type='application/zip',
data=self.archive2['data'],
CONTENT_LENGTH=self.archive2['length'],
HTTP_MD5SUM=self.archive2['md5sum'],
HTTP_SLUG='external-id',
HTTP_IN_PROGRESS=status_partial,
HTTP_CONTENT_DISPOSITION='attachment; filename=filename1.zip')
# then
assert response.status_code == status.HTTP_201_CREATED
response_content = parse_xml(BytesIO(response.content))
deposit_id = int(response_content['deposit_id'])
return deposit_id
def create_deposit_archive_with_archive(self, archive_extension):
# we create the holding archive to a given extension
archive = create_arborescence_archive(
self.root_path, 'archive1', 'file1', b'some content in file',
extension=archive_extension)
# now we create an archive holding the first created archive
invalid_archive = create_archive_with_archive(
self.root_path, 'invalid.tar.gz', archive)
# we deposit it
response = self.client.post(
reverse(COL_IRI, args=[self.collection.name]),
content_type='application/x-tar',
data=invalid_archive['data'],
CONTENT_LENGTH=invalid_archive['length'],
HTTP_MD5SUM=invalid_archive['md5sum'],
HTTP_SLUG='external-id',
HTTP_IN_PROGRESS=False,
HTTP_CONTENT_DISPOSITION='attachment; filename=%s' % (
invalid_archive['name'], ))
# then
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
response_content = parse_xml(BytesIO(response.content))
_status = response_content['deposit_status']
self.assertEqual(_status, DEPOSIT_STATUS_DEPOSITED)
deposit_id = int(response_content['deposit_id'])
return deposit_id
def update_binary_deposit(self, deposit_id, status_partial=False):
# update existing deposit with atom entry metadata
response = self.client.post(
reverse(EDIT_SE_IRI, args=[self.collection.name, deposit_id]),
content_type='application/atom+xml;type=entry',
data=self.codemeta_entry_data1,
HTTP_SLUG='external-id',
HTTP_IN_PROGRESS=status_partial)
# then
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
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'])
return deposit_id
@pytest.mark.fs
class BasicTestCase(TestCase):
"""Mixin intended for data setup purposes (user, collection, etc...)
"""
def setUp(self):
"""Define the test client and other test variables."""
super().setUp()
# expanding diffs in tests
self.maxDiff = None
# basic minimum test data
_name = 'hal'
_provider_url = 'https://hal-test.archives-ouvertes.fr/'
_domain = 'archives-ouvertes.fr/'
# set collection up
_collection = DepositCollection(name=_name)
_collection.save()
# set user/client up
_client = DepositClient.objects.create_user(username=_name,
password=_name,
provider_url=_provider_url,
domain=_domain)
_client.collections = [_collection.id]
_client.last_name = _name
_client.save()
self.collection = _collection
self.user = _client
self.username = _name
self.userpass = _name
def tearDown(self):
super().tearDown()
# Clean up uploaded files in temporary directory (tests have
# their own media root folder)
if os.path.exists(MEDIA_ROOT):
for d in os.listdir(MEDIA_ROOT):
shutil.rmtree(os.path.join(MEDIA_ROOT, d))
class WithAuthTestCase(TestCase):
"""Mixin intended for testing the api with basic authentication.
"""
def setUp(self):
super().setUp()
_token = '%s:%s' % (self.username, self.userpass)
token = base64.b64encode(_token.encode('utf-8'))
authorization = 'Basic %s' % token.decode('utf-8')
self.client.credentials(HTTP_AUTHORIZATION=authorization)
def tearDown(self):
super().tearDown()
self.client.credentials()
class CommonCreationRoutine(TestCase):
"""Mixin class to share initialization routine.
cf:
`class`:test_deposit_update.DepositReplaceExistingDataTest
`class`:test_deposit_update.DepositUpdateDepositWithNewDataTest
`class`:test_deposit_update.DepositUpdateFailuresTest
`class`:test_deposit_delete.DepositDeleteTest
"""
def setUp(self):
super().setUp()
self.atom_entry_data0 = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom">
<external_identifier>some-external-id</external_identifier>
<url>https://hal-test.archives-ouvertes.fr/some-external-id</url>
<author>some awesome author</author>
</entry>"""
self.atom_entry_data1 = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom">
<author>another one</author>
<author>no one</author>
</entry>"""
self.atom_entry_data2 = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom">
<title>Awesome Compiler</title>
<id>urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a</id>
<external_identifier>1785io25c695</external_identifier>
<updated>2017-10-07T15:17:08Z</updated>
<author>some awesome author</author>
<url>https://hal-test.archives-ouvertes.fr/id</url>
</entry>"""
self.codemeta_entry_data0 = b"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom"
xmlns:codemeta="https://doi.org/10.5063/SCHEMA/CODEMETA-2.0">
<title>Awesome Compiler</title>
<url>https://hal-test.archives-ouvertes.fr/1785io25c695</url>
<id>urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a</id>
<external_identifier>1785io25c695</external_identifier>
<updated>2017-10-07T15:17:08Z</updated>
<author>some awesome author</author>
<codemeta:description>description</codemeta:description>
<codemeta:keywords>key-word 1</codemeta:keywords>
</entry>"""
self.codemeta_entry_data1 = b"""<?xml version="1.0" encoding="utf-8"?>
<entry xmlns="http://www.w3.org/2005/Atom"
xmlns:codemeta="https://doi.org/10.5063/SCHEMA/CODEMETA-2.0">
<title>Composing a Web of Audio Applications</title>
<client>hal</client>
<id>hal-01243065</id>
<external_identifier>hal-01243065</external_identifier>
<codemeta:url>https://hal-test.archives-ouvertes.fr/hal-01243065</codemeta:url>
<codemeta:applicationCategory>test</codemeta:applicationCategory>
<codemeta:keywords>DSP programming,Web</codemeta:keywords>
<codemeta:dateCreated>2017-05-03T16:08:47+02:00</codemeta:dateCreated>
<codemeta:description>this is the description</codemeta:description>
<codemeta:version>1</codemeta:version>
<codemeta:runtimePlatform>phpstorm</codemeta:runtimePlatform>
<codemeta:developmentStatus>stable</codemeta:developmentStatus>
<codemeta:programmingLanguage>php</codemeta:programmingLanguage>
<codemeta:programmingLanguage>python</codemeta:programmingLanguage>
<codemeta:programmingLanguage>C</codemeta:programmingLanguage>
<codemeta:license>
<codemeta:name>GNU General Public License v3.0 only</codemeta:name>
</codemeta:license>
<codemeta:license>
<codemeta:name>CeCILL Free Software License Agreement v1.1</codemeta:name>
</codemeta:license>
<author>
<name>HAL</name>
<email>hal@ccsd.cnrs.fr</email>
</author>
<codemeta:author>
<codemeta:name>Morane Gruenpeter</codemeta:name>
</codemeta:author>
</entry>"""
def create_deposit_with_invalid_archive(self,
external_id='some-external-id-1'):
url = reverse(COL_IRI, args=[self.collection.name])
data = b'some data which is clearly not a zip file'
md5sum = hashlib.md5(data).hexdigest()
# when
response = self.client.post(
url,
content_type='application/zip', # as zip
data=data,
# + headers
CONTENT_LENGTH=len(data),
# other headers needs HTTP_ prefix to be taken into account
HTTP_SLUG=external_id,
HTTP_CONTENT_MD5=md5sum,
HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip',
HTTP_CONTENT_DISPOSITION='attachment; filename=filename0')
response_content = parse_xml(BytesIO(response.content))
deposit_id = int(response_content['deposit_id'])
return deposit_id
def create_deposit_with_status(
self, status,
external_id='some-external-id-1',
swh_id=None,
swh_id_context=None,
swh_anchor_id=None,
swh_anchor_id_context=None,
status_detail=None):
# 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
# test context ('rejected' for example). Update in place the
# deposit with such status to permit some further tests.
deposit = Deposit.objects.get(pk=deposit_id)
if status == DEPOSIT_STATUS_REJECTED:
deposit.status_detail = status_detail
deposit.status = status
if swh_id:
deposit.swh_id = swh_id
if swh_id_context:
deposit.swh_id_context = swh_id_context
if swh_anchor_id:
deposit.swh_anchor_id = swh_anchor_id
if swh_anchor_id_context:
deposit.swh_anchor_id_context = swh_anchor_id_context
deposit.save()
return deposit_id
def create_simple_deposit_partial(self, external_id='some-external-id'):
"""Create a simple deposit (1 request) in `partial` state and returns
its new identifier.
Returns:
deposit id
"""
response = self.client.post(
reverse(COL_IRI, args=[self.collection.name]),
content_type='application/atom+xml;type=entry',
data=self.atom_entry_data0,
HTTP_SLUG=external_id,
HTTP_IN_PROGRESS='true')
assert response.status_code == status.HTTP_201_CREATED
response_content = parse_xml(BytesIO(response.content))
deposit_id = int(response_content['deposit_id'])
return deposit_id
def create_deposit_partial_with_data_in_args(self, data):
"""Create a simple deposit (1 request) in `partial` state with the data
or metadata as an argument and returns its new identifier.
Args:
data: atom entry
Returns:
deposit id
"""
response = self.client.post(
reverse(COL_IRI, args=[self.collection.name]),
content_type='application/atom+xml;type=entry',
data=data,
HTTP_SLUG='external-id',
HTTP_IN_PROGRESS='true')
assert response.status_code == status.HTTP_201_CREATED
response_content = parse_xml(BytesIO(response.content))
deposit_id = int(response_content['deposit_id'])
return deposit_id
def _update_deposit_with_status(self, deposit_id, status_partial=False):
"""Add to a given deposit another archive and update its current
status to `deposited` (by default).
Returns:
deposit id
"""
# when
response = self.client.post(
reverse(EDIT_SE_IRI, args=[self.collection.name, deposit_id]),
content_type='application/atom+xml;type=entry',
data=self.atom_entry_data1,
HTTP_SLUG='external-id',
HTTP_IN_PROGRESS=status_partial)
# then
assert response.status_code == status.HTTP_201_CREATED
return deposit_id
def create_deposit_ready(self, external_id='some-external-id'):
"""Create a complex deposit (2 requests) in status `deposited`.
"""
deposit_id = self.create_simple_deposit_partial(
external_id=external_id)
deposit_id = self._update_deposit_with_status(deposit_id)
return deposit_id
def create_deposit_partial(self, external_id='some-external-id'):
"""Create a complex deposit (2 requests) in status `partial`.
"""
deposit_id = self.create_simple_deposit_partial(
external_id=external_id)
deposit_id = self._update_deposit_with_status(
deposit_id, status_partial=True)
return deposit_id
def add_metadata_to_deposit(self, deposit_id, status_partial=False):
"""Add metadata to deposit.
"""
# when
response = self.client.post(
reverse(EDIT_SE_IRI, args=[self.collection.name, deposit_id]),
content_type='application/atom+xml;type=entry',
data=self.codemeta_entry_data1,
HTTP_SLUG='external-id',
HTTP_IN_PROGRESS=status_partial)
assert response.status_code == status.HTTP_201_CREATED
# then
deposit = Deposit.objects.get(pk=deposit_id)
assert deposit is not None
deposit_requests = DepositRequest.objects.filter(deposit=deposit)
assert deposit_requests is not []
for dr in deposit_requests:
if dr.type == 'metadata':
assert deposit_requests[0].metadata is not {}
return deposit_id
diff --git a/swh/deposit/tests/loader/test_checker.py b/swh/deposit/tests/loader/test_checker.py
index d1721d64..6b45b4c2 100644
--- a/swh/deposit/tests/loader/test_checker.py
+++ b/swh/deposit/tests/loader/test_checker.py
@@ -1,68 +1,68 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# 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
from rest_framework.test import APITestCase
from swh.deposit.models import Deposit
from swh.deposit.config import PRIVATE_CHECK_DEPOSIT, DEPOSIT_STATUS_VERIFIED
from swh.deposit.config import DEPOSIT_STATUS_REJECTED
from swh.deposit.loader.checker import DepositChecker
-from django.core.urlresolvers import reverse
+from django.urls import reverse
from .common import SWHDepositTestClient, CLIENT_TEST_CONFIG
from ..common import BasicTestCase, WithAuthTestCase, CommonCreationRoutine
from ..common import FileSystemCreationRoutine
class DepositCheckerScenarioTest(APITestCase, WithAuthTestCase,
BasicTestCase, CommonCreationRoutine,
FileSystemCreationRoutine):
def setUp(self):
super().setUp()
# 2. Sets a basic client which accesses the test data
checker_client = SWHDepositTestClient(client=self.client,
config=CLIENT_TEST_CONFIG)
# 3. setup loader with no persistence and that client
self.checker = DepositChecker(client=checker_client)
def test_check_deposit_ready(self):
"""Check on a valid 'deposited' deposit should result in 'verified'
"""
# 1. create a deposit with archive and metadata
deposit_id = self.create_simple_binary_deposit()
deposit_id = self.update_binary_deposit(deposit_id,
status_partial=False)
args = [self.collection.name, deposit_id]
deposit_check_url = reverse(PRIVATE_CHECK_DEPOSIT, args=args)
# when
actual_result = self.checker.check(deposit_check_url=deposit_check_url)
# then
deposit = Deposit.objects.get(pk=deposit_id)
self.assertEqual(deposit.status, DEPOSIT_STATUS_VERIFIED)
self.assertEqual(actual_result, {'status': 'eventful'})
def test_check_deposit_rejected(self):
"""Check on invalid 'deposited' deposit should result in 'rejected'
"""
# 1. create a deposit with archive and metadata
deposit_id = self.create_deposit_with_invalid_archive()
args = [self.collection.name, deposit_id]
deposit_check_url = reverse(PRIVATE_CHECK_DEPOSIT, args=args)
# when
actual_result = self.checker.check(deposit_check_url=deposit_check_url)
# then
deposit = Deposit.objects.get(pk=deposit_id)
self.assertEqual(deposit.status, DEPOSIT_STATUS_REJECTED)
self.assertEqual(actual_result, {'status': 'eventful'})
diff --git a/swh/deposit/tests/loader/test_loader.py b/swh/deposit/tests/loader/test_loader.py
index e8cea274..3e03db67 100644
--- a/swh/deposit/tests/loader/test_loader.py
+++ b/swh/deposit/tests/loader/test_loader.py
@@ -1,169 +1,169 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# 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 os
import unittest
import shutil
import pytest
from rest_framework.test import APITestCase
from swh.model import hashutil
from swh.deposit.models import Deposit
from swh.deposit.loader import loader
from swh.deposit.config import (
PRIVATE_GET_RAW_CONTENT, PRIVATE_GET_DEPOSIT_METADATA, PRIVATE_PUT_DEPOSIT
)
-from django.core.urlresolvers import reverse
+from django.urls import reverse
from swh.loader.core.tests import BaseLoaderStorageTest
from .common import SWHDepositTestClient, CLIENT_TEST_CONFIG
from .. import TEST_LOADER_CONFIG
from ..common import (BasicTestCase, WithAuthTestCase,
CommonCreationRoutine,
FileSystemCreationRoutine)
class TestLoaderUtils(unittest.TestCase):
def assertRevisionsOk(self, expected_revisions): # noqa: N802
"""Check the loader's revisions match the expected revisions.
Expects self.loader to be instantiated and ready to be
inspected (meaning the loading took place).
Args:
expected_revisions (dict): Dict with key revision id,
value the targeted directory id.
"""
# The last revision being the one used later to start back from
for rev in self.loader.state['revision']:
rev_id = hashutil.hash_to_hex(rev['id'])
directory_id = hashutil.hash_to_hex(rev['directory'])
self.assertEqual(expected_revisions[rev_id], directory_id)
@pytest.mark.fs
class DepositLoaderScenarioTest(APITestCase, WithAuthTestCase,
BasicTestCase, CommonCreationRoutine,
FileSystemCreationRoutine, TestLoaderUtils,
BaseLoaderStorageTest):
def setUp(self):
super().setUp()
# create the extraction dir used by the loader
os.makedirs(TEST_LOADER_CONFIG['extraction_dir'], exist_ok=True)
# 1. create a deposit with archive and metadata
self.deposit_id = self.create_simple_binary_deposit()
# 2. Sets a basic client which accesses the test data
loader_client = SWHDepositTestClient(self.client,
config=CLIENT_TEST_CONFIG)
# 3. setup loader with that client
self.loader = loader.DepositLoader(client=loader_client)
self.storage = self.loader.storage
def tearDown(self):
super().tearDown()
shutil.rmtree(TEST_LOADER_CONFIG['extraction_dir'])
def test_inject_deposit_ready(self):
"""Load a deposit which is ready
"""
args = [self.collection.name, self.deposit_id]
archive_url = reverse(PRIVATE_GET_RAW_CONTENT, args=args)
deposit_meta_url = reverse(PRIVATE_GET_DEPOSIT_METADATA, args=args)
deposit_update_url = reverse(PRIVATE_PUT_DEPOSIT, args=args)
# when
res = self.loader.load(archive_url=archive_url,
deposit_meta_url=deposit_meta_url,
deposit_update_url=deposit_update_url)
# then
self.assertEqual(res['status'], 'eventful', res)
self.assertCountContents(1)
self.assertCountDirectories(1)
self.assertCountRevisions(1)
self.assertCountReleases(0)
self.assertCountSnapshots(1)
def test_inject_deposit_verify_metadata(self):
"""Load a deposit with metadata, test metadata integrity
"""
self.deposit_metadata_id = self.add_metadata_to_deposit(
self.deposit_id)
args = [self.collection.name, self.deposit_metadata_id]
archive_url = reverse(PRIVATE_GET_RAW_CONTENT, args=args)
deposit_meta_url = reverse(PRIVATE_GET_DEPOSIT_METADATA, args=args)
deposit_update_url = reverse(PRIVATE_PUT_DEPOSIT, args=args)
# when
self.loader.load(archive_url=archive_url,
deposit_meta_url=deposit_meta_url,
deposit_update_url=deposit_update_url)
# then
self.assertCountContents(1)
self.assertCountDirectories(1)
self.assertCountRevisions(1)
self.assertCountReleases(0)
self.assertCountSnapshots(1)
codemeta = 'codemeta:'
origin_url = 'https://hal-test.archives-ouvertes.fr/hal-01243065'
expected_origin_metadata = {
'@xmlns': 'http://www.w3.org/2005/Atom',
'@xmlns:codemeta': 'https://doi.org/10.5063/SCHEMA/CODEMETA-2.0',
'author': {
'email': 'hal@ccsd.cnrs.fr',
'name': 'HAL'
},
codemeta + 'url': origin_url,
codemeta + 'runtimePlatform': 'phpstorm',
codemeta + 'license': [
{
codemeta + 'name': 'GNU General Public License v3.0 only'
},
{
codemeta + 'name': 'CeCILL Free Software License Agreement v1.1' # noqa
}
],
codemeta + 'author': {
codemeta + 'name': 'Morane Gruenpeter'
},
codemeta + 'programmingLanguage': ['php', 'python', 'C'],
codemeta + 'applicationCategory': 'test',
codemeta + 'dateCreated': '2017-05-03T16:08:47+02:00',
codemeta + 'version': '1',
'external_identifier': 'hal-01243065',
'title': 'Composing a Web of Audio Applications',
codemeta + 'description': 'this is the description',
'id': 'hal-01243065',
'client': 'hal',
codemeta + 'keywords': 'DSP programming,Web',
codemeta + 'developmentStatus': 'stable'
}
self.assertOriginMetadataContains('deposit', origin_url,
expected_origin_metadata)
deposit = Deposit.objects.get(pk=self.deposit_id)
self.assertRegex(deposit.swh_id, r'^swh:1:dir:.*')
self.assertEqual(deposit.swh_id_context, '%s;origin=%s' % (
deposit.swh_id, origin_url
))
self.assertRegex(deposit.swh_anchor_id, r'^swh:1:rev:.*')
self.assertEqual(deposit.swh_anchor_id_context, '%s;origin=%s' % (
deposit.swh_anchor_id, origin_url
))

File Metadata

Mime Type
text/x-diff
Expires
Sat, Jun 21, 5:29 PM (1 w, 4 d ago)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
3266714

Event Timeline

About · Developer information · Legal