Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/tests/api/test_deposit_private_check.py
# Copyright (C) 2017-2019 The Software Heritage developers | # Copyright (C) 2017-2019 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
from django.urls import reverse | from django.urls import reverse | ||||
import pytest | |||||
from rest_framework import status | from rest_framework import status | ||||
from swh.deposit.config import ( | from swh.deposit.config import ( | ||||
DEPOSIT_STATUS_VERIFIED, PRIVATE_CHECK_DEPOSIT, | DEPOSIT_STATUS_VERIFIED, PRIVATE_CHECK_DEPOSIT, | ||||
DEPOSIT_STATUS_DEPOSITED, DEPOSIT_STATUS_REJECTED, COL_IRI | DEPOSIT_STATUS_DEPOSITED, DEPOSIT_STATUS_REJECTED, COL_IRI | ||||
) | ) | ||||
from swh.deposit.api.private.deposit_check import ( | from swh.deposit.api.private.deposit_check import ( | ||||
MANDATORY_ARCHIVE_INVALID, MANDATORY_FIELDS_MISSING, | MANDATORY_ARCHIVE_INVALID, MANDATORY_FIELDS_MISSING, | ||||
Show All 13 Lines | |||||
def private_check_url_endpoints(collection, deposit): | def private_check_url_endpoints(collection, deposit): | ||||
"""There are 2 endpoints to check (one with collection, one without)""" | """There are 2 endpoints to check (one with collection, one without)""" | ||||
return [ | return [ | ||||
reverse(PRIVATE_CHECK_DEPOSIT, args=[collection.name, deposit.id]), | reverse(PRIVATE_CHECK_DEPOSIT, args=[collection.name, deposit.id]), | ||||
reverse(PRIVATE_CHECK_DEPOSIT_NC, args=[deposit.id]) | reverse(PRIVATE_CHECK_DEPOSIT_NC, args=[deposit.id]) | ||||
] | ] | ||||
@pytest.mark.parametrize( | |||||
"extension", ['zip', 'tar', 'tar.gz', 'tar.bz2', 'tar.xz']) | |||||
def test_deposit_ok( | def test_deposit_ok( | ||||
authenticated_client, deposit_collection, ready_deposit_ok): | authenticated_client, deposit_collection, ready_deposit_ok, extension): | ||||
"""Proper deposit should succeed the checks (-> status ready) | """Proper deposit should succeed the checks (-> status ready) | ||||
""" | """ | ||||
deposit = ready_deposit_ok | deposit = ready_deposit_ok | ||||
for url in private_check_url_endpoints(deposit_collection, deposit): | for url in private_check_url_endpoints(deposit_collection, deposit): | ||||
response = authenticated_client.get(url) | response = authenticated_client.get(url) | ||||
assert response.status_code == status.HTTP_200_OK | assert response.status_code == status.HTTP_200_OK | ||||
data = response.json() | data = response.json() | ||||
assert data['status'] == DEPOSIT_STATUS_VERIFIED | assert data['status'] == DEPOSIT_STATUS_VERIFIED | ||||
deposit = Deposit.objects.get(pk=deposit.id) | deposit = Deposit.objects.get(pk=deposit.id) | ||||
assert deposit.status == DEPOSIT_STATUS_VERIFIED | assert deposit.status == DEPOSIT_STATUS_VERIFIED | ||||
deposit.status = DEPOSIT_STATUS_DEPOSITED | deposit.status = DEPOSIT_STATUS_DEPOSITED | ||||
deposit.save() | deposit.save() | ||||
@pytest.mark.parametrize( | |||||
"extension", ['zip', 'tar', 'tar.gz', 'tar.bz2', 'tar.xz']) | |||||
def test_deposit_invalid_tarball( | def test_deposit_invalid_tarball( | ||||
tmp_path, authenticated_client, deposit_collection): | tmp_path, authenticated_client, deposit_collection, extension): | ||||
"""Deposit with tarball (of 1 tarball) should fail the checks: rejected | """Deposit with tarball (of 1 tarball) should fail the checks: rejected | ||||
""" | """ | ||||
for archive_extension in ['zip', 'tar', 'tar.gz', 'tar.bz2', 'tar.xz']: | |||||
deposit = create_deposit_archive_with_archive( | deposit = create_deposit_archive_with_archive( | ||||
tmp_path, archive_extension, | tmp_path, extension, | ||||
authenticated_client, | authenticated_client, | ||||
deposit_collection.name) | deposit_collection.name) | ||||
for url in private_check_url_endpoints(deposit_collection, deposit): | for url in private_check_url_endpoints(deposit_collection, deposit): | ||||
response = authenticated_client.get(url) | response = authenticated_client.get(url) | ||||
assert response.status_code == status.HTTP_200_OK | assert response.status_code == status.HTTP_200_OK | ||||
data = response.json() | data = response.json() | ||||
assert data['status'] == DEPOSIT_STATUS_REJECTED | assert data['status'] == DEPOSIT_STATUS_REJECTED | ||||
details = data['details'] | details = data['details'] | ||||
# archive checks failure | # archive checks failure | ||||
assert len(details['archive']) == 1 | assert len(details['archive']) == 1 | ||||
assert details['archive'][0]['summary'] == \ | assert details['archive'][0]['summary'] == \ | ||||
MANDATORY_ARCHIVE_INVALID | MANDATORY_ARCHIVE_INVALID | ||||
deposit = Deposit.objects.get(pk=deposit.id) | deposit = Deposit.objects.get(pk=deposit.id) | ||||
assert deposit.status == DEPOSIT_STATUS_REJECTED | assert deposit.status == DEPOSIT_STATUS_REJECTED | ||||
def test_deposit_ko_missing_tarball( | def test_deposit_ko_missing_tarball( | ||||
authenticated_client, deposit_collection, ready_deposit_only_metadata): | authenticated_client, deposit_collection, ready_deposit_only_metadata): | ||||
"""Deposit without archive should fail the checks: rejected | """Deposit without archive should fail the checks: rejected | ||||
""" | """ | ||||
deposit = ready_deposit_only_metadata | deposit = ready_deposit_only_metadata | ||||
▲ Show 20 Lines • Show All 178 Lines • Show Last 20 Lines |