diff --git a/swh/deposit/api/deposit_status.py b/swh/deposit/api/deposit_status.py index e212f0ed..4795b9a0 100644 --- a/swh/deposit/api/deposit_status.py +++ b/swh/deposit/api/deposit_status.py @@ -1,120 +1,120 @@ # Copyright (C) 2017-2018 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 rest_framework import status from .common import SWHBaseDeposit from ..errors import NOT_FOUND, make_error_response from ..errors import make_error_response_from_dict from ..models import DEPOSIT_STATUS_DETAIL, Deposit def convert_status_detail(status_detail): """Given a status_detail dict, transforms it into a human readable string. Dict has the following form (all first level keys are optional): { 'url': { 'summary': , 'fields': }, 'metadata': [{ 'summary': , 'fields': , }], 'archive': [{ 'summary': , 'fields': , }] } Args: status_detail (dict): Returns: Status detail as inlined string. """ if not status_detail: return None msg = [] for key in ['metadata', 'archive']: _detail = status_detail.get(key) if _detail: for data in _detail: suffix_msg = '' fields = data.get('fields') if fields: - suffix_msg = ' (%s)' % ', '.join(fields) + suffix_msg = ' (%s)' % ', '.join(map(str, fields)) msg.append('- %s%s\n' % (data['summary'], suffix_msg)) _detail = status_detail.get('url') if _detail: fields = _detail.get('fields') suffix_msg = '' if fields: - suffix_msg = ' (%s)' % ', '.join(fields) + suffix_msg = ' (%s)' % ', '.join(map(str, fields)) msg.append('- %s%s\n' % (_detail['summary'], suffix_msg)) if not msg: return None return ''.join(msg) class SWHDepositStatus(SWHBaseDeposit): """Deposit status. What's known as 'State IRI' in the sword specification. HTTP verbs supported: GET """ def get(self, req, collection_name, deposit_id, format=None): checks = self.checks(req, collection_name, deposit_id) if 'error' in checks: return make_error_response_from_dict(req, checks['error']) try: deposit = Deposit.objects.get(pk=deposit_id) if deposit.collection.name != collection_name: raise Deposit.DoesNotExist except Deposit.DoesNotExist: return make_error_response( req, NOT_FOUND, 'deposit %s does not belong to collection %s' % ( deposit_id, collection_name)) status_detail = convert_status_detail(deposit.status_detail) if not status_detail: status_detail = DEPOSIT_STATUS_DETAIL[deposit.status] context = { 'deposit_id': deposit.id, 'status': deposit.status, 'status_detail': status_detail, 'swh_id': None, 'swh_id_context': None, 'swh_anchor_id': None, 'swh_anchor_id_context': None, } if deposit.swh_id: context['swh_id'] = deposit.swh_id if deposit.swh_id_context: context['swh_id_context'] = deposit.swh_id_context if deposit.swh_anchor_id: context['swh_anchor_id'] = deposit.swh_anchor_id if deposit.swh_anchor_id_context: context['swh_anchor_id_context'] = deposit.swh_anchor_id_context return render(req, 'deposit/status.xml', context=context, content_type='application/xml', status=status.HTTP_200_OK) diff --git a/swh/deposit/tests/api/test_deposit_status.py b/swh/deposit/tests/api/test_deposit_status.py index abbe07ff..fca43909 100644 --- a/swh/deposit/tests/api/test_deposit_status.py +++ b/swh/deposit/tests/api/test_deposit_status.py @@ -1,242 +1,277 @@ # Copyright (C) 2017-2018 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 io import BytesIO from nose.tools import istest from rest_framework import status from rest_framework.test import APITestCase from swh.deposit.api.deposit_status import convert_status_detail 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 """ @istest def 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]) @istest def 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) @istest def 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) @istest def 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) @istest def convert_status_detail_empty(self): actual_status_detail = convert_status_detail({}) self.assertIsNone(actual_status_detail) actual_status_detail = convert_status_detail({'dummy-keys': []}) self.assertIsNone(actual_status_detail) actual_status_detail = convert_status_detail(None) self.assertIsNone(actual_status_detail) @istest def convert_status_detail(self): status_detail = { 'url': { 'summary': "At least one url field must be compatible with the client\'s domain name. The following url fields failed the check", # noqa 'fields': ['blahurl', 'testurl'], }, 'metadata': [ { 'summary': 'Mandatory fields missing', 'fields': ['url', 'title'], }, { 'summary': 'Alternate fields missing', 'fields': ['name or title', 'url or badurl'] } ], 'archive': [{ 'summary': 'Unreadable archive', 'fields': ['1'], }], } expected_status_detail = '''- Mandatory fields missing (url, title) - Alternate fields missing (name or title, url or badurl) - Unreadable archive (1) - At least one url field must be compatible with the client's domain name. The following url fields failed the check (blahurl, testurl) ''' # noqa actual_status_detail = convert_status_detail(status_detail) self.assertEqual(actual_status_detail, expected_status_detail) @istest def convert_status_detail_2(self): status_detail = { 'url': { 'summary': 'At least one compatible url field. Failed', 'fields': ['testurl'], }, 'metadata': [ { 'summary': 'Mandatory fields missing', 'fields': ['name'], }, ], 'archive': [ { 'summary': 'Invalid archive', 'fields': ['2'], }, { 'summary': 'Unsupported archive', 'fields': ['1'], } ], } expected_status_detail = '''- Mandatory fields missing (name) - Invalid archive (2) - Unsupported archive (1) - At least one compatible url field. Failed (testurl) ''' actual_status_detail = convert_status_detail(status_detail) self.assertEqual(actual_status_detail, expected_status_detail) @istest def convert_status_detail_3(self): status_detail = { 'url': { 'summary': 'At least one compatible url field', }, } expected_status_detail = '- At least one compatible url field\n' actual_status_detail = convert_status_detail(status_detail) self.assertEqual(actual_status_detail, expected_status_detail) + @istest + def convert_status_detail_edge_case(self): + 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], + } + ], + } + + expected_status_detail = '''- Mandatory fields missing (9, 10, 1.212) +- Invalid archive (3) +- Unsupported archive (2) +- At least one compatible url field. Failed (testurl) +''' + + actual_status_detail = convert_status_detail(status_detail) + + self.assertEqual(actual_status_detail, expected_status_detail) + @istest def 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)