diff --git a/swh/deposit/tests/api/conftest.py b/swh/deposit/tests/api/conftest.py index 5234a40d..20819c52 100644 --- a/swh/deposit/tests/api/conftest.py +++ b/swh/deposit/tests/api/conftest.py @@ -1,95 +1,108 @@ # 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.config import ( + DEPOSIT_STATUS_DEPOSITED, COL_IRI, DEPOSIT_STATUS_VERIFIED +) 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_verified(partial_deposit_with_metadata): + """Returns a deposit ready for checks (it will pass the checks). + + """ + deposit = partial_deposit_with_metadata + deposit.status = DEPOSIT_STATUS_VERIFIED + 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_update_status.py b/swh/deposit/tests/api/test_deposit_private_update_status.py index f3b38b3f..cc347f86 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,140 @@ # 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 swh.deposit.tests.common import BasicTestCase +from swh.deposit.config import ( + PRIVATE_PUT_DEPOSIT, DEPOSIT_STATUS_VERIFIED, DEPOSIT_STATUS_LOAD_SUCCESS +) -class UpdateDepositStatusTest(APITestCase, BasicTestCase): - """Update the deposit's status scenario +PRIVATE_PUT_DEPOSIT_NC = PRIVATE_PUT_DEPOSIT + '-nc' - """ - 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 private_check_url_endpoints(collection, deposit): + """There are 2 endpoints to check (one with collection, one without)""" + return [ + reverse(PRIVATE_PUT_DEPOSIT, args=[collection.name, deposit.id]), + reverse(PRIVATE_PUT_DEPOSIT_NC, args=[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) +def test_update_deposit_status( + authenticated_client, deposit_collection, ready_deposit_verified): + """Existing status for update should return a 204 response + """ + deposit = ready_deposit_verified + for url in private_check_url_endpoints(deposit_collection, deposit): possible_status = set(DEPOSIT_STATUS_DETAIL.keys()) - set( [DEPOSIT_STATUS_LOAD_SUCCESS]) for _status in possible_status: - response = self.client.put( + response = authenticated_client.put( url, content_type='application/json', data=json.dumps({'status': _status})) - self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT) + assert response.status_code == status.HTTP_204_NO_CONTENT - deposit = Deposit.objects.get(pk=self.deposit.id) - self.assertEqual(deposit.status, _status) + deposit = Deposit.objects.get(pk=deposit.id) + assert deposit.status == _status - def test_update_deposit_status_with_info(self): - """Existing status for update with info should return a 204 response + deposit.status = DEPOSIT_STATUS_VERIFIED + deposit.save() # hack the same deposit - """ - url = self.private_deposit_url(self.deposit.id) +def test_update_deposit_status_with_info( + authenticated_client, deposit_collection, ready_deposit_verified): + """Existing status for update with info should return a 204 response + + """ + deposit = ready_deposit_verified + for url in private_check_url_endpoints(deposit_collection, deposit): 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( + response = authenticated_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) + assert response.status_code == status.HTTP_204_NO_CONTENT + + deposit = Deposit.objects.get(pk=deposit.id) + assert deposit.status == expected_status + assert deposit.swh_id == expected_swh_id + assert deposit.swh_id_context == expected_swh_id_context + assert deposit.swh_anchor_id == expected_swh_anchor_id + assert deposit.swh_anchor_id_context == expected_swh_anchor_id_context - 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) + deposit.swh_id = None + deposit.swh_id_context = None + deposit.swh_anchor_id = None + deposit.swh_anchor_id_context = None + deposit.status = DEPOSIT_STATUS_VERIFIED + deposit.save() - 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) +def test_update_deposit_status_will_fail_with_unknown_status( + authenticated_client, deposit_collection, ready_deposit_verified): + """Unknown status for update should return a 400 response - response = self.client.put( + """ + deposit = ready_deposit_verified + for url in private_check_url_endpoints(deposit_collection, deposit): + response = authenticated_client.put( url, content_type='application/json', data=json.dumps({'status': 'unknown'})) - self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + assert 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) +def test_update_deposit_status_will_fail_with_no_status_key( + authenticated_client, deposit_collection, ready_deposit_verified): + """No status provided for update should return a 400 response - response = self.client.put( + """ + deposit = ready_deposit_verified + for url in private_check_url_endpoints(deposit_collection, deposit): + response = authenticated_client.put( url, content_type='application/json', data=json.dumps({'something': 'something'})) - self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + assert 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) +def test_update_deposit_status_success_without_swh_id_fail( + authenticated_client, deposit_collection, ready_deposit_verified): + """Providing successful status without swh_id should return a 400 - response = self.client.put( + """ + deposit = ready_deposit_verified + for url in private_check_url_endpoints(deposit_collection, deposit): + response = authenticated_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]) + assert response.status_code == status.HTTP_400_BAD_REQUEST