Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/tests/api/test_deposit_check.py
# Copyright (C) 2017-2018 The Software Heritage developers | # Copyright (C) 2017-2018 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
import unittest | import unittest | ||||
from django.core.urlresolvers import reverse | from django.core.urlresolvers import reverse | ||||
from nose.tools import istest | |||||
from nose.plugins.attrib import attr | from nose.plugins.attrib import attr | ||||
from rest_framework import status | from rest_framework import status | ||||
from rest_framework.test import APITestCase | from rest_framework.test import APITestCase | ||||
from swh.deposit.config import ( | from swh.deposit.config import ( | ||||
DEPOSIT_STATUS_VERIFIED, PRIVATE_CHECK_DEPOSIT, | DEPOSIT_STATUS_VERIFIED, PRIVATE_CHECK_DEPOSIT, | ||||
DEPOSIT_STATUS_DEPOSITED, DEPOSIT_STATUS_REJECTED | DEPOSIT_STATUS_DEPOSITED, DEPOSIT_STATUS_REJECTED | ||||
) | ) | ||||
Show All 14 Lines | class CheckDepositTest(APITestCase, WithAuthTestCase, | ||||
BasicTestCase, CommonCreationRoutine, | BasicTestCase, CommonCreationRoutine, | ||||
FileSystemCreationRoutine): | FileSystemCreationRoutine): | ||||
"""Check deposit endpoints. | """Check deposit endpoints. | ||||
""" | """ | ||||
def setUp(self): | def setUp(self): | ||||
super().setUp() | super().setUp() | ||||
@istest | def test_deposit_ok(self): | ||||
def deposit_ok(self): | |||||
"""Proper deposit should succeed the checks (-> status ready) | """Proper deposit should succeed the checks (-> status ready) | ||||
""" | """ | ||||
deposit_id = self.create_simple_binary_deposit(status_partial=True) | deposit_id = self.create_simple_binary_deposit(status_partial=True) | ||||
deposit_id = self.update_binary_deposit(deposit_id, | deposit_id = self.update_binary_deposit(deposit_id, | ||||
status_partial=False) | status_partial=False) | ||||
deposit = Deposit.objects.get(pk=deposit_id) | deposit = Deposit.objects.get(pk=deposit_id) | ||||
self.assertEquals(deposit.status, DEPOSIT_STATUS_DEPOSITED) | self.assertEquals(deposit.status, DEPOSIT_STATUS_DEPOSITED) | ||||
url = reverse(PRIVATE_CHECK_DEPOSIT, | url = reverse(PRIVATE_CHECK_DEPOSIT, | ||||
args=[self.collection.name, deposit.id]) | args=[self.collection.name, deposit.id]) | ||||
response = self.client.get(url) | response = self.client.get(url) | ||||
self.assertEqual(response.status_code, status.HTTP_200_OK) | self.assertEqual(response.status_code, status.HTTP_200_OK) | ||||
data = response.json() | data = response.json() | ||||
self.assertEqual(data['status'], DEPOSIT_STATUS_VERIFIED) | self.assertEqual(data['status'], DEPOSIT_STATUS_VERIFIED) | ||||
deposit = Deposit.objects.get(pk=deposit.id) | deposit = Deposit.objects.get(pk=deposit.id) | ||||
self.assertEquals(deposit.status, DEPOSIT_STATUS_VERIFIED) | self.assertEquals(deposit.status, DEPOSIT_STATUS_VERIFIED) | ||||
@istest | def test_deposit_invalid_tarball(self): | ||||
def deposit_invalid_tarball(self): | |||||
"""Deposit with tarball (of 1 tarball) should fail the checks: rejected | """Deposit with tarball (of 1 tarball) should fail the checks: rejected | ||||
""" | """ | ||||
for archive_extension in ['zip', 'tar', 'tar.gz', 'tar.bz2', 'tar.xz']: | for archive_extension in ['zip', 'tar', 'tar.gz', 'tar.bz2', 'tar.xz']: | ||||
deposit_id = self.create_deposit_archive_with_archive( | deposit_id = self.create_deposit_archive_with_archive( | ||||
archive_extension) | archive_extension) | ||||
deposit = Deposit.objects.get(pk=deposit_id) | deposit = Deposit.objects.get(pk=deposit_id) | ||||
Show All 11 Lines | def test_deposit_invalid_tarball(self): | ||||
# archive checks failure | # archive checks failure | ||||
self.assertEqual(len(details['archive']), 1) | self.assertEqual(len(details['archive']), 1) | ||||
self.assertEqual(details['archive'][0]['summary'], | self.assertEqual(details['archive'][0]['summary'], | ||||
MANDATORY_ARCHIVE_INVALID) | MANDATORY_ARCHIVE_INVALID) | ||||
deposit = Deposit.objects.get(pk=deposit.id) | deposit = Deposit.objects.get(pk=deposit.id) | ||||
self.assertEquals(deposit.status, DEPOSIT_STATUS_REJECTED) | self.assertEquals(deposit.status, DEPOSIT_STATUS_REJECTED) | ||||
@istest | def test_deposit_ko_missing_tarball(self): | ||||
def deposit_ko_missing_tarball(self): | |||||
"""Deposit without archive should fail the checks: rejected | """Deposit without archive should fail the checks: rejected | ||||
""" | """ | ||||
deposit_id = self.create_deposit_ready() # no archive, only atom | deposit_id = self.create_deposit_ready() # no archive, only atom | ||||
deposit = Deposit.objects.get(pk=deposit_id) | deposit = Deposit.objects.get(pk=deposit_id) | ||||
self.assertEquals(DEPOSIT_STATUS_DEPOSITED, deposit.status) | self.assertEquals(DEPOSIT_STATUS_DEPOSITED, deposit.status) | ||||
url = reverse(PRIVATE_CHECK_DEPOSIT, | url = reverse(PRIVATE_CHECK_DEPOSIT, | ||||
args=[self.collection.name, deposit.id]) | args=[self.collection.name, deposit.id]) | ||||
response = self.client.get(url) | response = self.client.get(url) | ||||
self.assertEqual(response.status_code, status.HTTP_200_OK) | self.assertEqual(response.status_code, status.HTTP_200_OK) | ||||
data = response.json() | data = response.json() | ||||
self.assertEqual(data['status'], DEPOSIT_STATUS_REJECTED) | self.assertEqual(data['status'], DEPOSIT_STATUS_REJECTED) | ||||
details = data['details'] | details = data['details'] | ||||
# archive checks failure | # archive checks failure | ||||
self.assertEqual(len(details['archive']), 1) | self.assertEqual(len(details['archive']), 1) | ||||
self.assertEqual(details['archive'][0]['summary'], | self.assertEqual(details['archive'][0]['summary'], | ||||
MANDATORY_ARCHIVE_MISSING) | MANDATORY_ARCHIVE_MISSING) | ||||
deposit = Deposit.objects.get(pk=deposit.id) | deposit = Deposit.objects.get(pk=deposit.id) | ||||
self.assertEquals(deposit.status, DEPOSIT_STATUS_REJECTED) | self.assertEquals(deposit.status, DEPOSIT_STATUS_REJECTED) | ||||
@istest | def test_deposit_ko_unsupported_tarball(self): | ||||
def deposit_ko_unsupported_tarball(self): | |||||
"""Deposit with an unsupported tarball should fail the checks: rejected | """Deposit with an unsupported tarball should fail the checks: rejected | ||||
""" | """ | ||||
deposit_id = self.create_deposit_with_invalid_archive() | deposit_id = self.create_deposit_with_invalid_archive() | ||||
deposit = Deposit.objects.get(pk=deposit_id) | deposit = Deposit.objects.get(pk=deposit_id) | ||||
self.assertEquals(DEPOSIT_STATUS_DEPOSITED, deposit.status) | self.assertEquals(DEPOSIT_STATUS_DEPOSITED, deposit.status) | ||||
Show All 20 Lines | def test_deposit_ko_unsupported_tarball(self): | ||||
self.assertEqual(alternate['summary'], ALTERNATE_FIELDS_MISSING) | self.assertEqual(alternate['summary'], ALTERNATE_FIELDS_MISSING) | ||||
self.assertEqual(alternate['fields'], ['name or title']) | self.assertEqual(alternate['fields'], ['name or title']) | ||||
# url check failure | # url check failure | ||||
self.assertEqual(details['url']['summary'], INCOMPATIBLE_URL_FIELDS) | self.assertEqual(details['url']['summary'], INCOMPATIBLE_URL_FIELDS) | ||||
deposit = Deposit.objects.get(pk=deposit.id) | deposit = Deposit.objects.get(pk=deposit.id) | ||||
self.assertEquals(deposit.status, DEPOSIT_STATUS_REJECTED) | self.assertEquals(deposit.status, DEPOSIT_STATUS_REJECTED) | ||||
@istest | def test_check_deposit_metadata_ok(self): | ||||
def check_deposit_metadata_ok(self): | |||||
"""Proper deposit should succeed the checks (-> status ready) | """Proper deposit should succeed the checks (-> status ready) | ||||
with all **MUST** metadata | with all **MUST** metadata | ||||
using the codemeta metadata test set | using the codemeta metadata test set | ||||
""" | """ | ||||
deposit_id = self.create_simple_binary_deposit(status_partial=True) | deposit_id = self.create_simple_binary_deposit(status_partial=True) | ||||
deposit_id_metadata = self.add_metadata_to_deposit(deposit_id) | deposit_id_metadata = self.add_metadata_to_deposit(deposit_id) | ||||
self.assertEquals(deposit_id, deposit_id_metadata) | self.assertEquals(deposit_id, deposit_id_metadata) | ||||
Show All 10 Lines | def test_check_deposit_metadata_ok(self): | ||||
data = response.json() | data = response.json() | ||||
self.assertEqual(data['status'], DEPOSIT_STATUS_VERIFIED) | self.assertEqual(data['status'], DEPOSIT_STATUS_VERIFIED) | ||||
deposit = Deposit.objects.get(pk=deposit.id) | deposit = Deposit.objects.get(pk=deposit.id) | ||||
self.assertEquals(deposit.status, DEPOSIT_STATUS_VERIFIED) | self.assertEquals(deposit.status, DEPOSIT_STATUS_VERIFIED) | ||||
class CheckMetadata(unittest.TestCase, SWHChecksDeposit): | class CheckMetadata(unittest.TestCase, SWHChecksDeposit): | ||||
@istest | def test_check_metadata_ok(self): | ||||
def check_metadata_ok(self): | |||||
actual_check, detail = self._check_metadata({ | actual_check, detail = self._check_metadata({ | ||||
'url': 'something', | 'url': 'something', | ||||
'external_identifier': 'something-else', | 'external_identifier': 'something-else', | ||||
'name': 'foo', | 'name': 'foo', | ||||
'author': 'someone', | 'author': 'someone', | ||||
}) | }) | ||||
self.assertTrue(actual_check) | self.assertTrue(actual_check) | ||||
self.assertIsNone(detail) | self.assertIsNone(detail) | ||||
@istest | def test_check_metadata_ok2(self): | ||||
def check_metadata_ok2(self): | |||||
actual_check, detail = self._check_metadata({ | actual_check, detail = self._check_metadata({ | ||||
'url': 'something', | 'url': 'something', | ||||
'external_identifier': 'something-else', | 'external_identifier': 'something-else', | ||||
'title': 'bar', | 'title': 'bar', | ||||
'author': 'someone', | 'author': 'someone', | ||||
}) | }) | ||||
self.assertTrue(actual_check) | self.assertTrue(actual_check) | ||||
self.assertIsNone(detail) | self.assertIsNone(detail) | ||||
@istest | def test_check_metadata_ko(self): | ||||
def check_metadata_ko(self): | |||||
"""Missing optional field should be caught | """Missing optional field should be caught | ||||
""" | """ | ||||
actual_check, error_detail = self._check_metadata({ | actual_check, error_detail = self._check_metadata({ | ||||
'url': 'something', | 'url': 'something', | ||||
'external_identifier': 'something-else', | 'external_identifier': 'something-else', | ||||
'author': 'someone', | 'author': 'someone', | ||||
}) | }) | ||||
expected_error = { | expected_error = { | ||||
'metadata': [{ | 'metadata': [{ | ||||
'summary': 'Mandatory alternate fields are missing', | 'summary': 'Mandatory alternate fields are missing', | ||||
'fields': ['name or title'], | 'fields': ['name or title'], | ||||
}] | }] | ||||
} | } | ||||
self.assertFalse(actual_check) | self.assertFalse(actual_check) | ||||
self.assertEqual(error_detail, expected_error) | self.assertEqual(error_detail, expected_error) | ||||
@istest | def test_check_metadata_ko2(self): | ||||
def check_metadata_ko2(self): | |||||
"""Missing mandatory fields should be caught | """Missing mandatory fields should be caught | ||||
""" | """ | ||||
actual_check, error_detail = self._check_metadata({ | actual_check, error_detail = self._check_metadata({ | ||||
'url': 'something', | 'url': 'something', | ||||
'external_identifier': 'something-else', | 'external_identifier': 'something-else', | ||||
'title': 'foobar', | 'title': 'foobar', | ||||
}) | }) | ||||
Show All 10 Lines |