diff --git a/swh/deposit/tests/api/test_deposit_delete.py b/swh/deposit/tests/api/test_deposit_delete.py index 806d6325..ff6e066c 100644 --- a/swh/deposit/tests/api/test_deposit_delete.py +++ b/swh/deposit/tests/api/test_deposit_delete.py @@ -1,113 +1,121 @@ # 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 collections import defaultdict from django.urls import reverse - from rest_framework import status -from rest_framework.test import APITestCase +from typing import Mapping -from swh.deposit.config import EDIT_SE_IRI, EM_IRI, ARCHIVE_KEY, METADATA_KEY -from swh.deposit.config import DEPOSIT_STATUS_DEPOSITED +from swh.deposit.config import ( + EDIT_SE_IRI, EM_IRI, ARCHIVE_KEY, METADATA_KEY, + DEPOSIT_STATUS_DEPOSITED +) from swh.deposit.models import Deposit, DepositRequest -from ..common import BasicTestCase, WithAuthTestCase, CommonCreationRoutine - - -class DepositDeleteTest(APITestCase, WithAuthTestCase, BasicTestCase, - CommonCreationRoutine): - - def test_delete_archive_on_partial_deposit_works(self): - """Removing partial deposit's archive should return a 204 response - - """ - # given - deposit_id = self.create_deposit_partial() - deposit = Deposit.objects.get(pk=deposit_id) - deposit_requests = DepositRequest.objects.filter(deposit=deposit) - - self.assertEqual(len(deposit_requests), 2) - for dr in deposit_requests: - if dr.type == ARCHIVE_KEY: - continue - elif dr.type == METADATA_KEY: - continue - else: - self.fail('only archive and metadata type should exist ' - 'in this test context') - - # when - update_uri = reverse(EM_IRI, args=[self.collection.name, deposit_id]) - response = self.client.delete(update_uri) - # then - self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT) - - deposit = Deposit.objects.get(pk=deposit_id) - requests = list(DepositRequest.objects.filter(deposit=deposit)) - - self.assertEqual(len(requests), 2) - self.assertEqual(requests[0].type, 'metadata') - self.assertEqual(requests[1].type, 'metadata') - - def test_delete_archive_on_undefined_deposit_fails(self): - """Delete undefined deposit returns a 404 response - - """ - # when - update_uri = reverse(EM_IRI, args=[self.collection.name, 999]) - response = self.client.delete(update_uri) - # then - self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) - - def test_delete_archive_on_non_partial_deposit_fails(self): - """Delete !partial status deposit should return a 400 response""" - deposit_id = self.create_deposit_ready() - deposit = Deposit.objects.get(pk=deposit_id) - self.assertEqual(deposit.status, DEPOSIT_STATUS_DEPOSITED) - - # when - update_uri = reverse(EM_IRI, args=[self.collection.name, deposit_id]) - response = self.client.delete(update_uri) - # then - self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) - deposit = Deposit.objects.get(pk=deposit_id) - self.assertIsNotNone(deposit) - - def test_delete_partial_deposit_works(self): - """Delete deposit should return a 204 response - - """ - # given - deposit_id = self.create_simple_deposit_partial() - deposit = Deposit.objects.get(pk=deposit_id) - assert deposit.id == deposit_id - - # when - url = reverse(EDIT_SE_IRI, args=[self.collection.name, deposit_id]) - response = self.client.delete(url) - # then - self.assertEqual(response.status_code, - status.HTTP_204_NO_CONTENT) - deposit_requests = list(DepositRequest.objects.filter(deposit=deposit)) - self.assertEqual(deposit_requests, []) - deposits = list(Deposit.objects.filter(pk=deposit_id)) - self.assertEqual(deposits, []) - - def test_delete_on_edit_se_iri_cannot_delete_non_partial_deposit(self): - """Delete !partial deposit should return a 400 response - - """ - # given - deposit_id = self.create_deposit_ready() - deposit = Deposit.objects.get(pk=deposit_id) - assert deposit.id == deposit_id - - # when - url = reverse(EDIT_SE_IRI, args=[self.collection.name, deposit_id]) - response = self.client.delete(url) - # then - self.assertEqual(response.status_code, - status.HTTP_400_BAD_REQUEST) - deposit = Deposit.objects.get(pk=deposit_id) - self.assertIsNotNone(deposit) + + +def count_deposit_request_types(deposit_requests) -> Mapping[str, int]: + deposit_request_types = defaultdict(int) + for dr in deposit_requests: + deposit_request_types[dr.type] += 1 + return deposit_request_types + + +def test_delete_archive_on_partial_deposit_works( + authenticated_client, partial_deposit_with_metadata, + deposit_collection): + """Removing partial deposit's archive should return a 204 response + + """ + deposit_id = partial_deposit_with_metadata.id + deposit = Deposit.objects.get(pk=deposit_id) + deposit_requests = DepositRequest.objects.filter(deposit=deposit) + + # deposit request type: 'archive', 1 'metadata' + deposit_request_types = count_deposit_request_types(deposit_requests) + assert deposit_request_types == { + ARCHIVE_KEY: 1, + METADATA_KEY: 1 + } + + # when + update_uri = reverse(EM_IRI, args=[deposit_collection.name, deposit_id]) + response = authenticated_client.delete(update_uri) + + # then + assert response.status_code == status.HTTP_204_NO_CONTENT + + deposit = Deposit.objects.get(pk=deposit_id) + deposit_requests2 = DepositRequest.objects.filter(deposit=deposit) + + deposit_request_types = count_deposit_request_types(deposit_requests2) + assert deposit_request_types == { + METADATA_KEY: 1 + } + + +def test_delete_archive_on_undefined_deposit_fails( + authenticated_client, deposit_collection, sample_archive): + """Delete undefined deposit returns a 404 response + + """ + # when + update_uri = reverse(EM_IRI, args=[deposit_collection.name, 999]) + response = authenticated_client.delete(update_uri) + # then + assert response.status_code == status.HTTP_404_NOT_FOUND + + +def test_delete_non_partial_deposit( + authenticated_client, deposit_collection, deposited_deposit): + """Delete !partial status deposit should return a 400 response + + """ + deposit = deposited_deposit + assert deposit.status == DEPOSIT_STATUS_DEPOSITED + + # when + update_uri = reverse(EM_IRI, args=[deposit_collection.name, deposit.id]) + response = authenticated_client.delete(update_uri) + # then + assert response.status_code == status.HTTP_400_BAD_REQUEST + deposit = Deposit.objects.get(pk=deposit.id) + assert deposit is not None + + +def test_delete_partial_deposit( + authenticated_client, deposit_collection, partial_deposit): + """Delete deposit should return a 204 response + + """ + # given + deposit = partial_deposit + + # when + url = reverse(EDIT_SE_IRI, args=[deposit_collection.name, deposit.id]) + response = authenticated_client.delete(url) + # then + assert response.status_code == status.HTTP_204_NO_CONTENT + deposit_requests = list(DepositRequest.objects.filter(deposit=deposit)) + assert deposit_requests == [] + deposits = list(Deposit.objects.filter(pk=deposit.id)) + assert deposits == [] + + +def test_delete_on_edit_se_iri_cannot_delete_non_partial_deposit( + authenticated_client, deposit_collection, complete_deposit): + """Delete !partial deposit should return a 400 response + + """ + # given + deposit = complete_deposit + + # when + url = reverse(EDIT_SE_IRI, args=[deposit_collection.name, deposit.id]) + response = authenticated_client.delete(url) + # then + assert response.status_code == status.HTTP_400_BAD_REQUEST + deposit = Deposit.objects.get(pk=deposit.id) + assert deposit is not None diff --git a/swh/deposit/tests/conftest.py b/swh/deposit/tests/conftest.py index a5febe95..bd76ea38 100644 --- a/swh/deposit/tests/conftest.py +++ b/swh/deposit/tests/conftest.py @@ -1,230 +1,259 @@ # 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 # , STATE_IRI, from swh.scheduler.tests.conftest import * # noqa from swh.deposit.config import ( - COL_IRI, DEPOSIT_STATUS_DEPOSITED, DEPOSIT_STATUS_REJECTED, + COL_IRI, EDIT_SE_IRI, DEPOSIT_STATUS_DEPOSITED, DEPOSIT_STATUS_REJECTED, DEPOSIT_STATUS_PARTIAL, DEPOSIT_STATUS_LOAD_SUCCESS ) 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 @pytest.fixture def deposit_collection(db): from swh.deposit.models import DepositCollection collection_name = TEST_USER['collection']['name'] try: collection = DepositCollection._default_manager.get( name=collection_name) except DepositCollection.DoesNotExist: collection = DepositCollection(name=collection_name) collection.save() return 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): """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') # then assert response.status_code == status.HTTP_201_CREATED from swh.deposit.models import Deposit deposit = Deposit._default_manager.get(external_id=external_id) return deposit @pytest.fixture def deposited_deposit( sample_archive, deposit_collection, authenticated_client): - """Returns a deposit with status deposited. + """Returns a deposit with status 'deposited'. """ deposit = create_deposit( authenticated_client, deposit_collection.name, sample_archive, external_id='external-id-deposited') assert deposit.status == DEPOSIT_STATUS_DEPOSITED return deposit @pytest.fixture def rejected_deposit(sample_archive, deposit_collection, authenticated_client): - """Returns a deposit with status rejected. + """Returns a deposit with status 'rejected'. """ deposit = create_deposit( authenticated_client, deposit_collection.name, sample_archive, external_id='external-id-rejected') deposit.status = DEPOSIT_STATUS_REJECTED deposit.save() assert deposit.status == DEPOSIT_STATUS_REJECTED return deposit @pytest.fixture def partial_deposit(sample_archive, deposit_collection, authenticated_client): - """Returns a deposit with status rejected. + """Returns a deposit with status 'partial'. """ deposit = create_deposit( authenticated_client, deposit_collection.name, sample_archive, external_id='external-id-partial' ) deposit.status = DEPOSIT_STATUS_PARTIAL deposit.save() assert deposit.status == DEPOSIT_STATUS_PARTIAL return deposit +@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( + authenticated_client, deposit_collection.name, sample_archive, + external_id='external-id-partial' + ) + deposit.status = DEPOSIT_STATUS_PARTIAL + deposit.save() + assert deposit.status == DEPOSIT_STATUS_PARTIAL + + # 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