diff --git a/swh/deposit/tests/api/conftest.py b/swh/deposit/tests/api/conftest.py index c6c57749..5234a40d 100644 --- a/swh/deposit/tests/api/conftest.py +++ b/swh/deposit/tests/api/conftest.py @@ -1,32 +1,95 @@ # Copyright (C) 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 pytest +from django.urls import reverse from os import path, listdir from typing import Mapping +from swh.deposit.config import DEPOSIT_STATUS_DEPOSITED, COL_IRI +from swh.deposit.models import Deposit +from swh.deposit.parsers import parse_xml + +from swh.deposit.api.private.deposit_check import SWHChecksDeposit + @pytest.fixture def atom_dataset(datadir) -> Mapping[str, bytes]: """Compute the paths to atom files. Returns: Dict of atom name per content (bytes) """ atom_path = path.join(datadir, 'atom') - data = {} for filename in listdir(atom_path): filepath = path.join(atom_path, filename) with open(filepath, 'rb') as f: raw_content = f.read() # Keep the filename without extension atom_name = filename.split('.')[0] data[atom_name] = raw_content return data + + +@pytest.fixture +def ready_deposit_ok(partial_deposit_with_metadata): + """Returns a deposit ready for checks (it will pass the checks). + + """ + deposit = partial_deposit_with_metadata + deposit.status = DEPOSIT_STATUS_DEPOSITED + deposit.save() + return deposit + + +@pytest.fixture +def ready_deposit_only_metadata(partial_deposit_only_metadata): + """Deposit in status ready that will fail the checks (because missing + archive). + + """ + deposit = partial_deposit_only_metadata + deposit.status = DEPOSIT_STATUS_DEPOSITED + deposit.save() + return deposit + + +@pytest.fixture +def ready_deposit_invalid_archive(authenticated_client, deposit_collection): + url = reverse(COL_IRI, args=[deposit_collection.name]) + + data = b'some data which is clearly not a zip file' + md5sum = hashlib.md5(data).hexdigest() + + # when + response = authenticated_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-invalid', + HTTP_CONTENT_MD5=md5sum, + HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', + HTTP_CONTENT_DISPOSITION='attachment; filename=filename0') + + response_content = parse_xml(response.content) + deposit_id = int(response_content['deposit_id']) + deposit = Deposit.objects.get(pk=deposit_id) + deposit.status = DEPOSIT_STATUS_DEPOSITED + deposit.save() + return deposit + + +@pytest.fixture +def swh_checks_deposit(): + return SWHChecksDeposit() diff --git a/swh/deposit/tests/api/test_deposit_private_check.py b/swh/deposit/tests/api/test_deposit_private_check.py index 86d1d607..1c90113a 100644 --- a/swh/deposit/tests/api/test_deposit_private_check.py +++ b/swh/deposit/tests/api/test_deposit_private_check.py @@ -1,236 +1,263 @@ # 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.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 + DEPOSIT_STATUS_DEPOSITED, DEPOSIT_STATUS_REJECTED, COL_IRI ) from swh.deposit.api.private.deposit_check import ( - SWHChecksDeposit, MANDATORY_ARCHIVE_INVALID, - MANDATORY_FIELDS_MISSING, + MANDATORY_ARCHIVE_INVALID, MANDATORY_FIELDS_MISSING, MANDATORY_ARCHIVE_UNSUPPORTED, ALTERNATE_FIELDS_MISSING, MANDATORY_ARCHIVE_MISSING ) from swh.deposit.models import Deposit +from swh.deposit.parsers import parse_xml +from swh.deposit.tests.common import ( + create_arborescence_archive, create_archive_with_archive +) -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() +PRIVATE_CHECK_DEPOSIT_NC = PRIVATE_CHECK_DEPOSIT + '-nc' - def private_deposit_url(self, deposit_id): - return reverse(PRIVATE_CHECK_DEPOSIT, - args=[self.collection.name, deposit_id]) - def test_deposit_ok(self): - """Proper deposit should succeed the checks (-> status ready) +def private_check_url_endpoints(collection, deposit): + """There are 2 endpoints to check (one with collection, one without)""" + return [ + reverse(PRIVATE_CHECK_DEPOSIT, args=[collection.name, deposit.id]), + reverse(PRIVATE_CHECK_DEPOSIT_NC, args=[deposit.id]) + ] - """ - 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) +def test_deposit_ok( + authenticated_client, deposit_collection, ready_deposit_ok): + """Proper deposit should succeed the checks (-> status ready) - url = self.private_deposit_url(deposit.id) - response = self.client.get(url) + """ + deposit = ready_deposit_ok + for url in private_check_url_endpoints(deposit_collection, deposit): + response = authenticated_client.get(url) - self.assertEqual(response.status_code, status.HTTP_200_OK) + assert response.status_code == status.HTTP_200_OK data = response.json() - self.assertEqual(data['status'], DEPOSIT_STATUS_VERIFIED) + assert data['status'] == DEPOSIT_STATUS_VERIFIED deposit = Deposit.objects.get(pk=deposit.id) - self.assertEqual(deposit.status, DEPOSIT_STATUS_VERIFIED) + assert deposit.status == DEPOSIT_STATUS_VERIFIED - def test_deposit_invalid_tarball(self): - """Deposit with tarball (of 1 tarball) should fail the checks: rejected + deposit.status = DEPOSIT_STATUS_DEPOSITED + deposit.save() - """ - 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) +def test_deposit_invalid_tarball( + tmp_path, authenticated_client, deposit_collection): + """Deposit with tarball (of 1 tarball) should fail the checks: rejected - url = self.private_deposit_url(deposit.id) - response = self.client.get(url) - - self.assertEqual(response.status_code, status.HTTP_200_OK) + """ + for archive_extension in ['zip', 'tar', 'tar.gz', 'tar.bz2', 'tar.xz']: + deposit = create_deposit_archive_with_archive( + tmp_path, archive_extension, + authenticated_client, + deposit_collection.name) + for url in private_check_url_endpoints(deposit_collection, deposit): + response = authenticated_client.get(url) + assert response.status_code == status.HTTP_200_OK data = response.json() - self.assertEqual(data['status'], DEPOSIT_STATUS_REJECTED) + assert 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) + assert len(details['archive']) == 1 + assert details['archive'][0]['summary'] == \ + MANDATORY_ARCHIVE_INVALID deposit = Deposit.objects.get(pk=deposit.id) - self.assertEqual(deposit.status, DEPOSIT_STATUS_REJECTED) + assert 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) +def test_deposit_ko_missing_tarball( + authenticated_client, deposit_collection, ready_deposit_only_metadata): + """Deposit without archive should fail the checks: rejected - url = self.private_deposit_url(deposit.id) - response = self.client.get(url) + """ + deposit = ready_deposit_only_metadata + assert deposit.status == DEPOSIT_STATUS_DEPOSITED + + for url in private_check_url_endpoints(deposit_collection, deposit): + response = authenticated_client.get(url) - self.assertEqual(response.status_code, status.HTTP_200_OK) + assert response.status_code == status.HTTP_200_OK data = response.json() - self.assertEqual(data['status'], DEPOSIT_STATUS_REJECTED) + assert 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) + assert len(details['archive']) == 1 + assert details['archive'][0]['summary'] == MANDATORY_ARCHIVE_MISSING deposit = Deposit.objects.get(pk=deposit.id) - self.assertEqual(deposit.status, DEPOSIT_STATUS_REJECTED) + assert deposit.status == DEPOSIT_STATUS_REJECTED - def test_deposit_ko_unsupported_tarball(self): - """Deposit with an unsupported tarball should fail the checks: rejected + deposit.status = DEPOSIT_STATUS_DEPOSITED + deposit.save() - """ - deposit_id = self.create_deposit_with_invalid_archive() - deposit = Deposit.objects.get(pk=deposit_id) - self.assertEqual(DEPOSIT_STATUS_DEPOSITED, deposit.status) +def test_deposit_ko_unsupported_tarball( + tmp_path, authenticated_client, deposit_collection, + ready_deposit_invalid_archive): + """Deposit with an unsupported tarball should fail the checks: rejected + + """ + deposit = ready_deposit_invalid_archive + assert DEPOSIT_STATUS_DEPOSITED == deposit.status - url = self.private_deposit_url(deposit.id) - response = self.client.get(url) + for url in private_check_url_endpoints(deposit_collection, deposit): + response = authenticated_client.get(url) - self.assertEqual(response.status_code, status.HTTP_200_OK) + assert response.status_code == status.HTTP_200_OK data = response.json() - self.assertEqual(data['status'], DEPOSIT_STATUS_REJECTED) + assert 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) + assert len(details['archive']) == 1 + assert details['archive'][0]['summary'] == \ + MANDATORY_ARCHIVE_UNSUPPORTED # metadata check failure - self.assertEqual(len(details['metadata']), 2) + assert len(details['metadata']) == 2 mandatory = details['metadata'][0] - self.assertEqual(mandatory['summary'], MANDATORY_FIELDS_MISSING) - self.assertEqual(set(mandatory['fields']), - set(['author'])) + assert mandatory['summary'] == MANDATORY_FIELDS_MISSING + assert set(mandatory['fields']) == set(['author']) alternate = details['metadata'][1] - self.assertEqual(alternate['summary'], ALTERNATE_FIELDS_MISSING) - self.assertEqual(alternate['fields'], ['name or title']) + assert alternate['summary'] == ALTERNATE_FIELDS_MISSING + assert alternate['fields'] == ['name or title'] deposit = Deposit.objects.get(pk=deposit.id) - self.assertEqual(deposit.status, DEPOSIT_STATUS_REJECTED) + assert deposit.status == DEPOSIT_STATUS_REJECTED - def test_check_deposit_metadata_ok(self): - """Proper deposit should succeed the checks (-> status ready) - with all **MUST** metadata + deposit.status = DEPOSIT_STATUS_DEPOSITED + deposit.save() - 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) +def test_check_deposit_metadata_ok( + authenticated_client, deposit_collection, ready_deposit_ok): + """Proper deposit should succeed the checks (-> status ready) + with all **MUST** metadata - url = self.private_deposit_url(deposit.id) + using the codemeta metadata test set + """ + deposit = ready_deposit_ok + assert deposit.status == DEPOSIT_STATUS_DEPOSITED - response = self.client.get(url) + for url in private_check_url_endpoints(deposit_collection, deposit): + response = authenticated_client.get(url) - self.assertEqual(response.status_code, status.HTTP_200_OK) + assert response.status_code == status.HTTP_200_OK data = response.json() - self.assertEqual(data['status'], DEPOSIT_STATUS_VERIFIED) + assert data['status'] == DEPOSIT_STATUS_VERIFIED deposit = Deposit.objects.get(pk=deposit.id) - self.assertEqual(deposit.status, DEPOSIT_STATUS_VERIFIED) - - -@pytest.mark.fs -class CheckDepositTest2(CheckDepositTest): - def private_deposit_url(self, deposit_id): - return reverse(PRIVATE_CHECK_DEPOSIT+'-nc', - args=[deposit_id]) - - -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) + assert deposit.status == DEPOSIT_STATUS_VERIFIED + + deposit.status = DEPOSIT_STATUS_DEPOSITED + deposit.save() + + +def test_check_metadata_ok(swh_checks_deposit): + actual_check, detail = swh_checks_deposit._check_metadata({ + 'url': 'something', + 'external_identifier': 'something-else', + 'name': 'foo', + 'author': 'someone', + }) + + assert actual_check is True + assert detail is None + + +def test_check_metadata_ok2(swh_checks_deposit): + actual_check, detail = swh_checks_deposit._check_metadata({ + 'url': 'something', + 'external_identifier': 'something-else', + 'title': 'bar', + 'author': 'someone', + }) + + assert actual_check is True + assert detail is None + + +def test_check_metadata_ko(swh_checks_deposit): + """Missing optional field should be caught + + """ + actual_check, error_detail = swh_checks_deposit._check_metadata({ + 'url': 'something', + 'external_identifier': 'something-else', + 'author': 'someone', + }) + + expected_error = { + 'metadata': [{ + 'summary': 'Mandatory alternate fields are missing', + 'fields': ['name or title'], + }] + } + assert actual_check is False + assert error_detail == expected_error + + +def test_check_metadata_ko2(swh_checks_deposit): + """Missing mandatory fields should be caught + + """ + actual_check, error_detail = swh_checks_deposit._check_metadata({ + 'url': 'something', + 'external_identifier': 'something-else', + 'title': 'foobar', + }) + + expected_error = { + 'metadata': [{ + 'summary': 'Mandatory fields are missing', + 'fields': ['author'], + }] + } + + assert actual_check is False + assert error_detail == expected_error + + +def create_deposit_archive_with_archive( + root_path, archive_extension, client, collection_name): + # we create the holding archive to a given extension + archive = create_arborescence_archive( + 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( + root_path, 'invalid.tar.gz', archive) + + # we deposit it + response = client.post( + reverse(COL_IRI, args=[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 + assert response.status_code == status.HTTP_201_CREATED + response_content = parse_xml(response.content) + deposit_status = response_content['deposit_status'] + assert deposit_status == DEPOSIT_STATUS_DEPOSITED + deposit_id = int(response_content['deposit_id']) + + deposit = Deposit.objects.get(pk=deposit_id) + assert DEPOSIT_STATUS_DEPOSITED == deposit.status + return deposit diff --git a/swh/deposit/tests/api/test_deposit_private_read_archive.py b/swh/deposit/tests/api/test_deposit_private_read_archive.py index b4ec2f41..e86e980d 100644 --- a/swh/deposit/tests/api/test_deposit_private_read_archive.py +++ b/swh/deposit/tests/api/test_deposit_private_read_archive.py @@ -1,98 +1,100 @@ # 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.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 +from swh.deposit.tests.common import ( + BasicTestCase, WithAuthTestCase, CommonCreationRoutine, + 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 private_deposit_url(self, deposit_id): return reverse(PRIVATE_GET_RAW_CONTENT, args=[self.collection.name, deposit_id]) 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 = self.private_deposit_url(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 = self.private_deposit_url(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']), []) @pytest.mark.fs class DepositReadArchivesTest2(DepositReadArchivesTest): def private_deposit_url(self, deposit_id): return reverse(PRIVATE_GET_RAW_CONTENT+'-nc', args=[deposit_id]) diff --git a/swh/deposit/tests/api/test_deposit_private_read_metadata.py b/swh/deposit/tests/api/test_deposit_private_read_metadata.py index e4110a5a..2ccc10f9 100644 --- a/swh/deposit/tests/api/test_deposit_private_read_metadata.py +++ b/swh/deposit/tests/api/test_deposit_private_read_metadata.py @@ -1,644 +1,646 @@ # 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.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 +from swh.deposit.config import SWH_PERSON +from swh.deposit.tests.common import ( + BasicTestCase, WithAuthTestCase, CommonCreationRoutine +) class DepositReadMetadataTest(APITestCase, WithAuthTestCase, BasicTestCase, CommonCreationRoutine): """Deposit access to read metadata information on deposit. """ def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.template_metadata = """ Composing a Web of Audio Applications hal hal-01243065 hal-01243065 https://hal-test.archives-ouvertes.fr/hal-01243065 test DSP programming this is the description 1 phpstorm stable php python C GNU General Public License v3.0 only CeCILL Free Software License Agreement v1.1 HAL hal@ccsd.cnrs.fr Morane Gruenpeter %s """ def private_deposit_url(self, deposit_id): return reverse(PRIVATE_GET_DEPOSIT_METADATA, args=[self.collection.name, deposit_id]) def test_read_metadata(self): """Private metadata read api to existing deposit should return metadata """ deposit_id = self.create_deposit_partial() url = self.private_deposit_url(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'], 'codemeta:dateCreated': '2017-10-07T15:17:08Z', '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': { 'timestamp': { 'seconds': 1507389428, 'microseconds': 0 }, 'offset': 0, 'negative_utc': False }, 'message': 'hal: Deposit %s in collection hal' % deposit_id, 'author': SWH_PERSON, 'committer': SWH_PERSON, 'date': { 'timestamp': { 'seconds': 1507389428, 'microseconds': 0 }, 'offset': 0, 'negative_utc': False }, 'metadata': { '@xmlns': ['http://www.w3.org/2005/Atom'], 'author': ['some awesome author', 'another one', 'no one'], 'external_identifier': 'some-external-id', 'codemeta:dateCreated': '2017-10-07T15:17:08Z', '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 = self.private_deposit_url(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'], 'codemeta:dateCreated': '2017-10-07T15:17:08Z', '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': { 'timestamp': { 'seconds': 1507389428, 'microseconds': 0 }, 'offset': 0, 'negative_utc': False }, 'committer_date': { 'timestamp': { 'seconds': 1507389428, 'microseconds': 0 }, 'offset': 0, 'negative_utc': False }, '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'], 'codemeta:dateCreated': '2017-10-07T15:17:08Z', '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_read_metadata_3(self): """date(Created|Published) provided, uses author/committer date """ # add metadata to the deposit with datePublished and dateCreated codemeta_entry_data = self.template_metadata % """ 2015-04-06T17:08:47+02:00 2017-05-03T16:08:47+02:00 """ deposit_id = self.create_deposit_partial_with_data_in_args( codemeta_entry_data) url = self.private_deposit_url(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_origin = { 'type': 'deposit', 'url': 'https://hal-test.archives-ouvertes.fr/hal-01243065' } expected_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' }, 'client': 'hal', 'codemeta:applicationCategory': 'test', 'codemeta:author': { 'codemeta:name': 'Morane Gruenpeter' }, 'codemeta:dateCreated': '2015-04-06T17:08:47+02:00', 'codemeta:datePublished': '2017-05-03T16:08:47+02:00', 'codemeta:description': 'this is the description', 'codemeta:developmentStatus': 'stable', 'codemeta:keywords': 'DSP programming', 'codemeta:license': [ { 'codemeta:name': 'GNU General Public License v3.0 only' }, { 'codemeta:name': 'CeCILL Free Software License Agreement v1.1' } ], 'codemeta:programmingLanguage': [ 'php', 'python', 'C' ], 'codemeta:runtimePlatform': 'phpstorm', 'codemeta:url': 'https://hal-test.archives-ouvertes.fr/hal-01243065', # noqa 'codemeta:version': '1', 'external_identifier': 'hal-01243065', 'id': 'hal-01243065', 'title': 'Composing a Web of Audio Applications' } expected_origin_metadata = { 'metadata': expected_metadata, 'provider': { 'metadata': {}, 'provider_name': 'hal', 'provider_type': 'deposit_client', 'provider_url': 'https://hal-test.archives-ouvertes.fr/' }, 'tool': { 'configuration': { 'sword_version': '2' }, 'name': 'swh-deposit', 'version': '0.0.1' } } expected_revision = { 'author': { 'email': 'robot@softwareheritage.org', 'fullname': 'Software Heritage', 'name': 'Software Heritage' }, 'committer': { 'email': 'robot@softwareheritage.org', 'fullname': 'Software Heritage', 'name': 'Software Heritage' }, 'committer_date': { 'negative_utc': False, 'offset': 120, 'timestamp': { 'microseconds': 0, 'seconds': 1493820527 } }, 'date': { 'negative_utc': False, 'offset': 120, 'timestamp': { 'microseconds': 0, 'seconds': 1428332927 } }, 'message': 'hal: Deposit %s in collection hal' % deposit_id, 'metadata': expected_metadata, 'synthetic': True, 'type': 'tar' } expected_meta = { 'branch_name': 'master', 'origin': expected_origin, 'origin_metadata': expected_origin_metadata, 'revision': expected_revision, } self.assertEqual(data, expected_meta) def test_read_metadata_4(self): """dateCreated/datePublished not provided, revision uses complete_date """ codemeta_entry_data = self.template_metadata % '' deposit_id = self.create_deposit_partial_with_data_in_args( codemeta_entry_data) # will use the deposit completed date as fallback date deposit = Deposit.objects.get(pk=deposit_id) deposit.complete_date = '2016-04-06' deposit.save() url = self.private_deposit_url(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_origin = { 'type': 'deposit', 'url': 'https://hal-test.archives-ouvertes.fr/hal-01243065' } expected_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' }, 'client': 'hal', 'codemeta:applicationCategory': 'test', 'codemeta:author': { 'codemeta:name': 'Morane Gruenpeter' }, 'codemeta:description': 'this is the description', 'codemeta:developmentStatus': 'stable', 'codemeta:keywords': 'DSP programming', 'codemeta:license': [ { 'codemeta:name': 'GNU General Public License v3.0 only' }, { 'codemeta:name': 'CeCILL Free Software License Agreement v1.1' } ], 'codemeta:programmingLanguage': [ 'php', 'python', 'C' ], 'codemeta:runtimePlatform': 'phpstorm', 'codemeta:url': 'https://hal-test.archives-ouvertes.fr/hal-01243065', # noqa 'codemeta:version': '1', 'external_identifier': 'hal-01243065', 'id': 'hal-01243065', 'title': 'Composing a Web of Audio Applications' } expected_origin_metadata = { 'metadata': expected_metadata, 'provider': { 'metadata': {}, 'provider_name': 'hal', 'provider_type': 'deposit_client', 'provider_url': 'https://hal-test.archives-ouvertes.fr/' }, 'tool': { 'configuration': { 'sword_version': '2' }, 'name': 'swh-deposit', 'version': '0.0.1' } } expected_revision = { 'author': { 'email': 'robot@softwareheritage.org', 'fullname': 'Software Heritage', 'name': 'Software Heritage' }, 'committer': { 'email': 'robot@softwareheritage.org', 'fullname': 'Software Heritage', 'name': 'Software Heritage' }, 'committer_date': { 'negative_utc': False, 'offset': 0, 'timestamp': { 'microseconds': 0, 'seconds': 1459900800 } }, 'date': { 'negative_utc': False, 'offset': 0, 'timestamp': { 'microseconds': 0, 'seconds': 1459900800 } }, 'message': 'hal: Deposit %s in collection hal' % deposit_id, 'metadata': expected_metadata, 'synthetic': True, 'type': 'tar' } expected_meta = { 'branch_name': 'master', 'origin': expected_origin, 'origin_metadata': expected_origin_metadata, 'revision': expected_revision, } self.assertEqual(data, expected_meta) def test_read_metadata_5(self): """dateCreated/datePublished provided, revision uses author/committer date If multiple dateCreated provided, the first occurrence (of dateCreated) is selected. If multiple datePublished provided, the first occurrence (of datePublished) is selected. """ # add metadata to the deposit with multiple datePublished/dateCreated codemeta_entry_data = self.template_metadata % """ 2015-04-06T17:08:47+02:00 2017-05-03T16:08:47+02:00 2016-04-06T17:08:47+02:00 2018-05-03T16:08:47+02:00 """ deposit_id = self.create_deposit_partial_with_data_in_args( codemeta_entry_data) url = self.private_deposit_url(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_origin = { 'type': 'deposit', 'url': 'https://hal-test.archives-ouvertes.fr/hal-01243065' } expected_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' }, 'client': 'hal', 'codemeta:applicationCategory': 'test', 'codemeta:author': { 'codemeta:name': 'Morane Gruenpeter' }, 'codemeta:dateCreated': [ '2015-04-06T17:08:47+02:00', '2016-04-06T17:08:47+02:00', ], 'codemeta:datePublished': [ '2017-05-03T16:08:47+02:00', '2018-05-03T16:08:47+02:00', ], 'codemeta:description': 'this is the description', 'codemeta:developmentStatus': 'stable', 'codemeta:keywords': 'DSP programming', 'codemeta:license': [ { 'codemeta:name': 'GNU General Public License v3.0 only' }, { 'codemeta:name': 'CeCILL Free Software License Agreement v1.1' } ], 'codemeta:programmingLanguage': [ 'php', 'python', 'C' ], 'codemeta:runtimePlatform': 'phpstorm', 'codemeta:url': 'https://hal-test.archives-ouvertes.fr/hal-01243065', # noqa 'codemeta:version': '1', 'external_identifier': 'hal-01243065', 'id': 'hal-01243065', 'title': 'Composing a Web of Audio Applications' } expected_origin_metadata = { 'metadata': expected_metadata, 'provider': { 'metadata': {}, 'provider_name': 'hal', 'provider_type': 'deposit_client', 'provider_url': 'https://hal-test.archives-ouvertes.fr/' }, 'tool': { 'configuration': { 'sword_version': '2' }, 'name': 'swh-deposit', 'version': '0.0.1' } } expected_revision = { 'author': { 'email': 'robot@softwareheritage.org', 'fullname': 'Software Heritage', 'name': 'Software Heritage' }, 'committer': { 'email': 'robot@softwareheritage.org', 'fullname': 'Software Heritage', 'name': 'Software Heritage' }, 'committer_date': { 'negative_utc': False, 'offset': 120, 'timestamp': { 'microseconds': 0, 'seconds': 1493820527 } }, 'date': { 'negative_utc': False, 'offset': 120, 'timestamp': { 'microseconds': 0, 'seconds': 1428332927 } }, 'message': 'hal: Deposit %s in collection hal' % deposit_id, 'metadata': expected_metadata, 'synthetic': True, 'type': 'tar' } expected_meta = { 'branch_name': 'master', 'origin': expected_origin, 'origin_metadata': expected_origin_metadata, 'revision': expected_revision, } 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 = self.private_deposit_url(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')) class DepositReadMetadataTest2(DepositReadMetadataTest): def private_deposit_url(self, deposit_id): return reverse(PRIVATE_GET_DEPOSIT_METADATA+'-nc', args=[deposit_id]) diff --git a/swh/deposit/tests/api/test_deposit_private_update_status.py b/swh/deposit/tests/api/test_deposit_private_update_status.py index 7e6185f3..f3b38b3f 100644 --- a/swh/deposit/tests/api/test_deposit_private_update_status.py +++ b/swh/deposit/tests/api/test_deposit_private_update_status.py @@ -1,134 +1,134 @@ # 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.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 +from swh.deposit.tests.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 private_deposit_url(self, deposit_id): return reverse(PRIVATE_PUT_DEPOSIT, args=[self.collection.name, deposit_id]) def test_update_deposit_status(self): """Existing status for update should return a 204 response """ url = self.private_deposit_url(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 = self.private_deposit_url(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 = self.private_deposit_url(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 = self.private_deposit_url(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 = self.private_deposit_url(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) class UpdateDepositStatusTest2(UpdateDepositStatusTest): def private_deposit_url(self, deposit_id): return reverse(PRIVATE_PUT_DEPOSIT+'-nc', args=[deposit_id]) diff --git a/swh/deposit/tests/conftest.py b/swh/deposit/tests/conftest.py index 71767d42..e2a60538 100644 --- a/swh/deposit/tests/conftest.py +++ b/swh/deposit/tests/conftest.py @@ -1,279 +1,306 @@ # Copyright (C) 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 pytest import psycopg2 from django.urls import reverse from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT from rest_framework import status from rest_framework.test import APIClient from typing import Mapping from swh.scheduler.tests.conftest import * # noqa - +from swh.deposit.parsers import parse_xml from swh.deposit.config import ( COL_IRI, EDIT_SE_IRI, DEPOSIT_STATUS_DEPOSITED, DEPOSIT_STATUS_REJECTED, DEPOSIT_STATUS_PARTIAL, DEPOSIT_STATUS_LOAD_SUCCESS, DEPOSIT_STATUS_LOAD_FAILURE ) from swh.deposit.tests.common import create_arborescence_archive TEST_USER = { 'username': 'test', 'password': 'password', 'email': 'test@example.org', 'provider_url': 'https://hal-test.archives-ouvertes.fr/', 'domain': 'archives-ouvertes.fr/', 'collection': { 'name': 'test' }, } def execute_sql(sql): """Execute sql to postgres db""" with psycopg2.connect(database='postgres') as conn: conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) cur = conn.cursor() cur.execute(sql) @pytest.hookimpl(tryfirst=True) def pytest_load_initial_conftests(early_config, parser, args): """This hook is done prior to django loading. Used to initialize the deposit's server db. """ import project.app.signals def prepare_db(*args, **kwargs): from django.conf import settings db_name = 'tests' print('before: %s' % settings.DATABASES) # work around db settings for django for k, v in [ ('ENGINE', 'django.db.backends.postgresql'), ('NAME', 'tests'), ('USER', postgresql_proc.user), # noqa ('HOST', postgresql_proc.host), # noqa ('PORT', postgresql_proc.port), # noqa ]: settings.DATABASES['default'][k] = v print('after: %s' % settings.DATABASES) execute_sql('DROP DATABASE IF EXISTS %s' % db_name) execute_sql('CREATE DATABASE %s TEMPLATE template0' % db_name) project.app.signals.something = prepare_db def create_deposit_collection(collection_name: str): """Create a deposit collection with name collection_name """ from swh.deposit.models import DepositCollection try: collection = DepositCollection._default_manager.get( name=collection_name) except DepositCollection.DoesNotExist: collection = DepositCollection(name=collection_name) collection.save() return collection def deposit_collection_factory( collection_name=TEST_USER['collection']['name']): @pytest.fixture def _deposit_collection(db, collection_name=collection_name): return create_deposit_collection(collection_name) return _deposit_collection deposit_collection = deposit_collection_factory() deposit_another_collection = deposit_collection_factory('another-collection') @pytest.fixture def deposit_user(db, deposit_collection): """Create/Return the test_user "test" """ from swh.deposit.models import DepositClient try: user = DepositClient._default_manager.get( username=TEST_USER['username']) except DepositClient.DoesNotExist: user = DepositClient._default_manager.create_user( username=TEST_USER['username'], email=TEST_USER['email'], password=TEST_USER['password'], provider_url=TEST_USER['provider_url'], domain=TEST_USER['domain'], ) user.collections = [deposit_collection.id] user.save() return user @pytest.fixture def client(): """Override pytest-django one which does not work for djangorestframework. """ return APIClient() # <- drf's client @pytest.yield_fixture def authenticated_client(client, deposit_user): """Returned a logged client """ _token = '%s:%s' % (deposit_user.username, TEST_USER['password']) token = base64.b64encode(_token.encode('utf-8')) authorization = 'Basic %s' % token.decode('utf-8') client.credentials(HTTP_AUTHORIZATION=authorization) yield client client.logout() @pytest.fixture def sample_archive(tmp_path): """Returns a sample archive """ tmp_path = str(tmp_path) # pytest version limitation in previous version archive = create_arborescence_archive( tmp_path, 'archive1', 'file1', b'some content in file') return archive def create_deposit( authenticated_client, collection_name: str, sample_archive, external_id: str, deposit_status=DEPOSIT_STATUS_DEPOSITED): """Create a skeleton shell deposit """ url = reverse(COL_IRI, args=[collection_name]) # when response = authenticated_client.post( url, content_type='application/zip', # as zip data=sample_archive['data'], # + headers CONTENT_LENGTH=sample_archive['length'], HTTP_SLUG=external_id, HTTP_CONTENT_MD5=sample_archive['md5sum'], HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip', HTTP_IN_PROGRESS='false', HTTP_CONTENT_DISPOSITION='attachment; filename=%s' % ( sample_archive['name'])) # then assert response.status_code == status.HTTP_201_CREATED from swh.deposit.models import Deposit deposit = Deposit._default_manager.get(external_id=external_id) if deposit.status != deposit_status: deposit.status = deposit_status deposit.save() assert deposit.status == deposit_status return deposit def create_binary_deposit( authenticated_client, collection_name: str, sample_archive, external_id: str, deposit_status: str = DEPOSIT_STATUS_DEPOSITED, atom_dataset: Mapping[str, bytes] = {}): """Create a deposit with both metadata and archive set. Then alters its status to `deposit_status`. """ deposit = create_deposit( authenticated_client, collection_name, sample_archive, external_id=external_id, deposit_status=DEPOSIT_STATUS_PARTIAL) response = authenticated_client.post( reverse(EDIT_SE_IRI, args=[collection_name, deposit.id]), content_type='application/atom+xml;type=entry', data=atom_dataset['entry-data0'] % deposit.external_id.encode('utf-8'), HTTP_SLUG=deposit.external_id, HTTP_IN_PROGRESS='true') assert response.status_code == status.HTTP_201_CREATED assert deposit.status == DEPOSIT_STATUS_PARTIAL from swh.deposit.models import Deposit deposit = Deposit._default_manager.get(pk=deposit.id) if deposit.status != deposit_status: deposit.status = deposit_status deposit.save() assert deposit.status == deposit_status return deposit def deposit_factory(deposit_status=DEPOSIT_STATUS_DEPOSITED): """Build deposit with a specific status """ @pytest.fixture() def _deposit(sample_archive, deposit_collection, authenticated_client, deposit_status=deposit_status): external_id = 'external-id-%s' % deposit_status return create_deposit( authenticated_client, deposit_collection.name, sample_archive, external_id=external_id, deposit_status=deposit_status ) return _deposit deposited_deposit = deposit_factory() rejected_deposit = deposit_factory(deposit_status=DEPOSIT_STATUS_REJECTED) partial_deposit = deposit_factory(deposit_status=DEPOSIT_STATUS_PARTIAL) completed_deposit = deposit_factory(deposit_status=DEPOSIT_STATUS_LOAD_SUCCESS) failed_deposit = deposit_factory(deposit_status=DEPOSIT_STATUS_LOAD_FAILURE) @pytest.fixture def partial_deposit_with_metadata( sample_archive, deposit_collection, authenticated_client, atom_dataset): """Returns deposit with archive and metadata provided, status 'partial' """ return create_binary_deposit( authenticated_client, deposit_collection.name, sample_archive, external_id='external-id-partial', deposit_status=DEPOSIT_STATUS_PARTIAL, atom_dataset=atom_dataset ) +@pytest.fixture +def partial_deposit_only_metadata( + deposit_collection, authenticated_client, + atom_dataset): + + response = authenticated_client.post( + reverse(COL_IRI, args=[deposit_collection.name]), + content_type='application/atom+xml;type=entry', + data=atom_dataset['entry-data1'], + HTTP_SLUG='external-id-partial', + HTTP_IN_PROGRESS=True) + + assert response.status_code == status.HTTP_201_CREATED + + response_content = parse_xml(response.content) + deposit_id = response_content['deposit_id'] + from swh.deposit.models import Deposit + deposit = Deposit._default_manager.get(pk=deposit_id) + assert deposit.status == DEPOSIT_STATUS_PARTIAL + return deposit + + @pytest.fixture def complete_deposit(sample_archive, deposit_collection, authenticated_client): """Returns a completed deposit (load success) """ deposit = create_deposit( authenticated_client, deposit_collection.name, sample_archive, external_id='external-id-complete', deposit_status=DEPOSIT_STATUS_LOAD_SUCCESS ) _swh_id_context = 'https://hal.archives-ouvertes.fr/hal-01727745' deposit.swh_id = 'swh:1:dir:42a13fc721c8716ff695d0d62fc851d641f3a12b' deposit.swh_id_context = '%s;%s' % ( deposit.swh_id, _swh_id_context) deposit.swh_anchor_id = \ 'swh:rev:1:548b3c0a2bb43e1fca191e24b5803ff6b3bc7c10' deposit.swh_anchor_id_context = '%s;%s' % ( deposit.swh_anchor_id, _swh_id_context) deposit.save() return deposit + + +@pytest.fixture() +def tmp_path(tmp_path): + return str(tmp_path) # issue with oldstable's pytest version