diff --git a/swh/deposit/tests/api/test_deposit_update.py b/swh/deposit/tests/api/test_deposit_update.py index 227c1a2d..ffc86cff 100644 --- a/swh/deposit/tests/api/test_deposit_update.py +++ b/swh/deposit/tests/api/test_deposit_update.py @@ -1,333 +1,383 @@ # 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, DepositRequest +from swh.deposit.models import Deposit, DepositRequest, DepositCollection from swh.deposit.config import EDIT_SE_IRI, EM_IRI +from swh.deposit.parsers import parse_xml -from ..common import BasicTestCase, WithAuthTestCase, CommonCreationRoutine -from ..common import FileSystemCreationRoutine, create_arborescence_archive +from swh.deposit.tests.common import create_arborescence_archive, check_archive -class DepositUpdateOrReplaceExistingDataTest( - APITestCase, WithAuthTestCase, BasicTestCase, - FileSystemCreationRoutine, CommonCreationRoutine): - """Try put/post (update/replace) query on EM_IRI +def test_replace_archive_to_deposit_is_possible( + tmp_path, partial_deposit, deposit_collection, authenticated_client, + sample_archive, atom_dataset): + """Replace all archive with another one should return a 204 response """ - def setUp(self): - super().setUp() + tmp_path = str(tmp_path) + # given + deposit = partial_deposit + requests = DepositRequest.objects.filter( + deposit=deposit, + type='archive') + + assert len(list(requests)) == 1 + check_archive(sample_archive['name'], requests[0].archive.name) + + # we have no metadata for that deposit + requests = list(DepositRequest.objects.filter( + deposit=deposit, type='metadata')) + assert len(requests) == 0 + + response = authenticated_client.post( + reverse(EDIT_SE_IRI, args=[deposit_collection.name, deposit.id]), + content_type='application/atom+xml;type=entry', + data=atom_dataset['entry-data1'], + HTTP_SLUG=deposit.external_id, + HTTP_IN_PROGRESS=True) + + requests = list(DepositRequest.objects.filter( + deposit=deposit, type='metadata')) + assert len(requests) == 1 + + update_uri = reverse(EM_IRI, args=[deposit_collection.name, deposit.id]) + external_id = 'some-external-id-1' + archive2 = create_arborescence_archive( + tmp_path, 'archive2', 'file2', b'some other content in file') + + response = authenticated_client.put( + 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_IN_PROGRESS='false', + HTTP_CONTENT_DISPOSITION='attachment; filename=%s' % ( + archive2['name'], )) + + assert response.status_code == status.HTTP_204_NO_CONTENT + + requests = DepositRequest.objects.filter( + deposit=deposit, + type='archive') + + assert len(list(requests)) == 1 + check_archive(archive2['name'], requests[0].archive.name) + + # check we did not touch the other parts + requests = list(DepositRequest.objects.filter( + deposit=deposit, type='metadata')) + assert len(requests) == 1 + + +def test_replace_metadata_to_deposit_is_possible( + tmp_path, authenticated_client, partial_deposit_with_metadata, + deposit_collection, atom_dataset): + """Replace all metadata with another one should return a 204 response - self.atom_entry_data1 = b""" - - bar -""" - - self.atom_entry_data1 = b""" - - bar -""" - - 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]) + """ + # given + deposit = partial_deposit_with_metadata + raw_metadata0 = atom_dataset['entry-data0'] % deposit.external_id.encode( + 'utf-8') + + requests_meta = DepositRequest.objects.filter( + deposit=deposit, + type='metadata') + assert len(requests_meta) == 1 + request_meta0 = requests_meta[0] + assert request_meta0.raw_metadata == raw_metadata0.decode('utf-8') + + requests_archive0 = DepositRequest.objects.filter( + deposit=deposit, type='archive') + assert len(requests_archive0) == 1 + + update_uri = reverse(EDIT_SE_IRI, args=[ + deposit_collection.name, deposit.id]) + + response = authenticated_client.put( + update_uri, + content_type='application/atom+xml;type=entry', + data=atom_dataset['entry-data1']) + + assert response.status_code == status.HTTP_204_NO_CONTENT + + requests_meta = DepositRequest.objects.filter( + deposit=deposit, + type='metadata') + + assert len(requests_meta) == 1 + request_meta1 = requests_meta[0] + raw_metadata1 = request_meta1.raw_metadata + assert raw_metadata1 == atom_dataset['entry-data1'].decode('utf-8') + assert raw_metadata0 != raw_metadata1 + assert request_meta0 != request_meta1 + + # check we did not touch the other parts + requests_archive1 = DepositRequest.objects.filter( + deposit=deposit, type='archive') + assert len(requests_archive1) == 1 + assert set(requests_archive0) == set(requests_archive1) + + +def test_add_archive_to_deposit_is_possible( + tmp_path, authenticated_client, deposit_collection, + partial_deposit_with_metadata, sample_archive): + """Add another archive to a deposit return a 201 response - external_id = 'some-external-id-1' + """ + tmp_path = str(tmp_path) + deposit = partial_deposit_with_metadata + + requests = DepositRequest.objects.filter( + deposit=deposit, + type='archive') + + assert len(requests) == 1 + check_archive(sample_archive['name'], requests[0].archive.name) + + requests_meta0 = DepositRequest.objects.filter( + deposit=deposit, type='metadata') + assert len(requests_meta0) == 1 + + update_uri = reverse(EM_IRI, args=[deposit_collection.name, deposit.id]) + + external_id = 'some-external-id-1' + archive2 = create_arborescence_archive( + tmp_path, 'archive2', 'file2', b'some other content in file') + + response = authenticated_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_IN_PROGRESS='false', + HTTP_CONTENT_DISPOSITION='attachment; filename=%s' % ( + archive2['name'],)) + + assert response.status_code == status.HTTP_201_CREATED + + requests = DepositRequest.objects.filter( + deposit=deposit, + type='archive').order_by('id') + + assert len(requests) == 2 + # first archive still exists + check_archive(sample_archive['name'], requests[0].archive.name) + # a new one was added + check_archive(archive2['name'], requests[1].archive.name) + + # check we did not touch the other parts + requests_meta1 = DepositRequest.objects.filter( + deposit=deposit, type='metadata') + assert len(requests_meta1) == 1 + assert set(requests_meta0) == set(requests_meta1) + + +def test_add_metadata_to_deposit_is_possible( + authenticated_client, deposit_collection, + partial_deposit_with_metadata, atom_dataset): + """Add metadata with another one should return a 204 response - 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) + """ + deposit = partial_deposit_with_metadata + requests = DepositRequest.objects.filter( + deposit=deposit, + type='metadata') - requests = list(DepositRequest.objects.filter( - deposit=deposit, - type='archive').order_by('id')) + assert len(requests) == 1 - 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']) + requests_archive0 = DepositRequest.objects.filter( + deposit=deposit, type='archive') + assert len(requests_archive0) == 1 - # check we did not touch the other parts - requests = list(DepositRequest.objects.filter( - deposit=deposit, type='metadata')) - self.assertEqual(len(requests), 0) + update_uri = reverse(EDIT_SE_IRI, args=[deposit_collection.name, + deposit.id]) - def test_add_metadata_to_deposit_is_possible(self): - """Add metadata with another one should return a 204 response + atom_entry = atom_dataset['entry-data1'] + response = authenticated_client.post( + update_uri, + content_type='application/atom+xml;type=entry', + data=atom_entry) - """ - # given - deposit_id = self.create_deposit_partial() + assert response.status_code == status.HTTP_201_CREATED - deposit = Deposit.objects.get(pk=deposit_id) - requests = DepositRequest.objects.filter( - deposit=deposit, - type='metadata') + requests = DepositRequest.objects.filter( + deposit=deposit, + type='metadata').order_by('id') - assert len(list(requests)) == 2 + assert len(requests) == 2 + expected_raw_meta0 = atom_dataset['entry-data0'] % ( + deposit.external_id.encode('utf-8')) + # a new one was added + assert requests[0].raw_metadata == expected_raw_meta0.decode('utf-8') + assert requests[1].raw_metadata == atom_entry.decode('utf-8') - requests = list(DepositRequest.objects.filter( - deposit=deposit, type='archive')) - assert len(requests) == 0 + # check we did not touch the other parts + requests_archive1 = DepositRequest.objects.filter( + deposit=deposit, type='archive') + assert len(requests_archive1) == 1 + assert set(requests_archive0) == set(requests_archive1) - 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) +def test_add_metadata_to_unknown_deposit( + deposit_collection, authenticated_client, atom_dataset): + """Replacing metadata to unknown deposit should return a 404 response - self.assertEqual(response.status_code, status.HTTP_201_CREATED) + """ + unknown_deposit_id = 1000 + try: + Deposit.objects.get(pk=unknown_deposit_id) + except Deposit.DoesNotExist: + assert True + + url = reverse(EDIT_SE_IRI, args=[deposit_collection, unknown_deposit_id]) + response = authenticated_client.post( + url, + content_type='application/atom+xml;type=entry', + data=atom_dataset['entry-data1']) + assert response.status_code == status.HTTP_404_NOT_FOUND + response_content = parse_xml(response.content) + assert 'Unknown collection name' in \ + response_content['sword:error']['summary'] + + +def test_add_metadata_to_unknown_collection( + partial_deposit, authenticated_client, atom_dataset): + """Replacing metadata to unknown deposit should return a 404 response - requests = DepositRequest.objects.filter( - deposit=deposit, - type='metadata').order_by('id') + """ + deposit = partial_deposit + unknown_collection_name = 'unknown-collection' + try: + DepositCollection.objects.get(name=unknown_collection_name) + except DepositCollection.DoesNotExist: + assert True + + url = reverse(EDIT_SE_IRI, args=[unknown_collection_name, deposit.id]) + response = authenticated_client.post( + url, + content_type='application/atom+xml;type=entry', + data=atom_dataset['entry-data1']) + assert response.status_code == status.HTTP_404_NOT_FOUND + response_content = parse_xml(response.content) + assert 'Unknown collection name' in \ + response_content['sword:error']['summary'] + + +def test_replace_metadata_to_unknown_deposit( + authenticated_client, deposit_collection, atom_dataset): + """Adding metadata to unknown deposit should return a 404 response - self.assertEqual(len(list(requests)), 3) - # a new one was added - self.assertEqual(requests[1].metadata['foobar'], 'bar') + """ + unknown_deposit_id = 998 + try: + Deposit.objects.get(pk=unknown_deposit_id) + except Deposit.DoesNotExist: + assert True + url = reverse(EDIT_SE_IRI, args=[ + deposit_collection.name, unknown_deposit_id]) + response = authenticated_client.put( + url, + content_type='application/atom+xml;type=entry', + data=atom_dataset['entry-data1']) + assert response.status_code == status.HTTP_404_NOT_FOUND + response_content = parse_xml(response.content) + assert 'Deposit with id %s does not exist' % unknown_deposit_id == \ + response_content['sword:error']['summary'] + + +def test_add_archive_to_unknown_deposit( + authenticated_client, deposit_collection, atom_dataset): + """Adding metadata to unknown deposit should return a 404 response - # check we did not touch the other parts - requests = list(DepositRequest.objects.filter( - deposit=deposit, type='archive')) - self.assertEqual(len(requests), 0) + """ + unknown_deposit_id = 997 + try: + Deposit.objects.get(pk=unknown_deposit_id) + except Deposit.DoesNotExist: + assert True + + url = reverse(EM_IRI, args=[deposit_collection.name, unknown_deposit_id]) + response = authenticated_client.post(url, + content_type='application/zip', + data=atom_dataset['entry-data1']) + assert response.status_code == status.HTTP_404_NOT_FOUND + response_content = parse_xml(response.content) + assert 'Deposit with id %s does not exist' % unknown_deposit_id == \ + response_content['sword:error']['summary'] + + +def test_replace_archive_to_unknown_deposit( + authenticated_client, deposit_collection, atom_dataset): + """Replacing archive to unknown deposit should return a 404 response + """ + unknown_deposit_id = 996 + try: + Deposit.objects.get(pk=unknown_deposit_id) + except Deposit.DoesNotExist: + assert True + + url = reverse(EM_IRI, args=[deposit_collection.name, unknown_deposit_id]) + response = authenticated_client.put( + url, + content_type='application/zip', + data=atom_dataset['entry-data1']) + assert response.status_code == status.HTTP_404_NOT_FOUND + response_content = parse_xml(response.content) + assert 'Deposit with id %s does not exist' % unknown_deposit_id == \ + response_content['sword:error']['summary'] + + +def test_post_metadata_to_em_iri_failure( + authenticated_client, deposit_collection, partial_deposit, + atom_dataset): + """Update (POST) archive with wrong content type should return 400 -class DepositUpdateFailuresTest(APITestCase, WithAuthTestCase, BasicTestCase, - CommonCreationRoutine): - """Failure scenario about add/replace (post/put) query on deposit. + """ + deposit = partial_deposit + update_uri = reverse(EM_IRI, args=[deposit_collection.name, deposit.id]) + response = authenticated_client.post( + update_uri, + content_type='application/x-gtar-compressed', + data=atom_dataset['entry-data1']) + assert response.status_code == status.HTTP_400_BAD_REQUEST + response_content = parse_xml(response.content) + msg = 'Packaging format supported is restricted to ' + \ + 'application/zip, application/x-tar' + assert msg == response_content['sword:error']['summary'] + + +def test_put_metadata_to_em_iri_failure( + authenticated_client, deposit_collection, partial_deposit, + atom_dataset): + """Update (PUT) archive with wrong content type should return 400 """ - 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') + # given + deposit = partial_deposit + # when + update_uri = reverse(EM_IRI, args=[deposit_collection.name, deposit.id]) + response = authenticated_client.put( + update_uri, + content_type='application/atom+xml;type=entry', + data=atom_dataset['entry-data1']) + # then + assert response.status_code == status.HTTP_400_BAD_REQUEST + response_content = parse_xml(response.content) + msg = 'Packaging format supported is restricted to ' + \ + 'application/zip, application/x-tar' + assert msg == response_content['sword:error']['summary'] diff --git a/swh/deposit/tests/conftest.py b/swh/deposit/tests/conftest.py index b816e5f5..71767d42 100644 --- a/swh/deposit/tests/conftest.py +++ b/swh/deposit/tests/conftest.py @@ -1,257 +1,279 @@ # 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 -# , STATE_IRI, from swh.scheduler.tests.conftest import * # noqa 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']): +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=filename0') + 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' """ - # deposit with one archive - deposit = create_deposit( + return create_binary_deposit( authenticated_client, deposit_collection.name, sample_archive, external_id='external-id-partial', - deposit_status=DEPOSIT_STATUS_PARTIAL + deposit_status=DEPOSIT_STATUS_PARTIAL, + atom_dataset=atom_dataset ) - # update the deposit with metadata - response = authenticated_client.post( - reverse(EDIT_SE_IRI, args=[deposit_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 - 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