diff --git a/swh/deposit/tests/api/test_deposit.py b/swh/deposit/tests/api/test_deposit.py
index 59b1179f..d5dfe69f 100644
--- a/swh/deposit/tests/api/test_deposit.py
+++ b/swh/deposit/tests/api/test_deposit.py
@@ -1,156 +1,189 @@
# 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
from django.urls import reverse
from io import BytesIO
from rest_framework import status
-from rest_framework.test import APITestCase
-from swh.deposit.config import COL_IRI, EDIT_SE_IRI, DEPOSIT_STATUS_REJECTED
-from swh.deposit.config import DEPOSIT_STATUS_PARTIAL
-from swh.deposit.config import DEPOSIT_STATUS_LOAD_SUCCESS
-from swh.deposit.config import DEPOSIT_STATUS_LOAD_FAILURE
-from swh.deposit.models import Deposit, DepositClient, DepositCollection
-from swh.deposit.parsers import parse_xml
+from swh.deposit.config import (
+ COL_IRI, EDIT_SE_IRI, DEPOSIT_STATUS_REJECTED,
+ DEPOSIT_STATUS_PARTIAL, DEPOSIT_STATUS_LOAD_SUCCESS,
+ DEPOSIT_STATUS_LOAD_FAILURE
+)
-from ..common import BasicTestCase, WithAuthTestCase, CommonCreationRoutine
+from swh.deposit.models import Deposit
+from swh.deposit.parsers import parse_xml
def test_deposit_post_will_fail_with_401(client):
"""Without authentication, endpoint refuses access with 401 response
"""
url = reverse(COL_IRI, args=['hal'])
+ response = client.post(url)
+ assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_access_to_another_user_collection_is_forbidden(
+ authenticated_client, deposit_another_collection, deposit_user):
+ """Access to another user collection should return a 403
+
+ """
+ coll2 = deposit_another_collection
+ url = reverse(COL_IRI, args=[coll2.name])
+ response = authenticated_client.post(url)
+ assert response.status_code == status.HTTP_403_FORBIDDEN
+ msg = 'Client %s cannot access collection %s' % (
+ deposit_user.username, coll2.name, )
+ assert msg in response.content.decode('utf-8')
+
+
+def test_delete_on_col_iri_not_supported(
+ authenticated_client, deposit_collection):
+ """Delete on col iri should return a 405 response
+
+ """
+ url = reverse(COL_IRI, args=[deposit_collection.name])
+ response = authenticated_client.delete(url)
+ assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED
+ assert 'DELETE method is not supported on this endpoint' in \
+ response.content.decode('utf-8')
+
+
+def create_deposit_with_rejection_status(
+ 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()
+ external_id = 'some-external-id-1'
# when
- response = client.post(url)
+ 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,
+ HTTP_CONTENT_MD5=md5sum,
+ HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip',
+ HTTP_CONTENT_DISPOSITION='attachment; filename=filename0')
+
+ assert response.status_code == status.HTTP_201_CREATED
+ response_content = parse_xml(BytesIO(response.content))
+ actual_state = response_content['deposit_status']
+ assert actual_state == DEPOSIT_STATUS_REJECTED
+
+
+def test_act_on_deposit_rejected_is_not_permitted(
+ authenticated_client, deposit_collection, rejected_deposit,
+ atom_dataset):
+ deposit = rejected_deposit
+
+ 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)
+
+ assert response.status_code == status.HTTP_400_BAD_REQUEST
+ msg = 'You can only act on deposit with status '%s'' % (
+ DEPOSIT_STATUS_PARTIAL, )
+ assert msg in response.content.decode('utf-8')
+
+
+def test_add_deposit_when_partial_makes_new_deposit(
+ authenticated_client, deposit_collection, partial_deposit,
+ atom_dataset):
+ """Posting deposit on collection when previous is partial makes new deposit
- # then
- assert response.status_code == status.HTTP_401_UNAUTHORIZED
+ """
+ deposit = partial_deposit
+ assert deposit.status == DEPOSIT_STATUS_PARTIAL
+
+ # adding a new deposit with the same external id
+ response = authenticated_client.post(
+ reverse(COL_IRI, args=[deposit_collection.name]),
+ content_type='application/atom+xml;type=entry',
+ data=atom_dataset['entry-data0'] % deposit.external_id.encode('utf-8'),
+ HTTP_SLUG=deposit.external_id
+ )
+
+ assert response.status_code == status.HTTP_201_CREATED
+ response_content = parse_xml(BytesIO(response.content))
+ deposit_id = response_content['deposit_id']
+
+ assert deposit_id != deposit.id # new deposit
+
+ new_deposit = Deposit.objects.get(pk=deposit_id)
+ assert new_deposit != deposit
+ assert new_deposit.parent is None
+
+
+def test_add_deposit_when_failed_makes_new_deposit_with_no_parent(
+ authenticated_client, deposit_collection, failed_deposit,
+ atom_dataset):
+ """Posting deposit on collection when deposit done makes new deposit with
+ parent
+
+ """
+ deposit = failed_deposit
+ assert deposit.status == DEPOSIT_STATUS_LOAD_FAILURE
+
+ # adding a new deposit with the same external id as a completed deposit
+ # creates the parenting chain
+ response = authenticated_client.post(
+ reverse(COL_IRI, args=[deposit_collection.name]),
+ content_type='application/atom+xml;type=entry',
+ data=atom_dataset['entry-data0'] % deposit.external_id.encode('utf-8'),
+ HTTP_SLUG=deposit.external_id)
+
+ assert response.status_code == status.HTTP_201_CREATED
+ response_content = parse_xml(BytesIO(response.content))
+ deposit_id = response_content['deposit_id']
+
+ assert deposit_id != deposit.id
+
+ new_deposit = Deposit.objects.get(pk=deposit_id)
+ assert new_deposit != deposit
+ assert new_deposit.parent is None
-class DepositFailuresTest(APITestCase, WithAuthTestCase, BasicTestCase,
- CommonCreationRoutine):
- """Deposit access are protected with basic authentication.
+def test_add_deposit_when_done_makes_new_deposit_with_parent_old_one(
+ authenticated_client, deposit_collection, completed_deposit,
+ atom_dataset):
+ """Posting deposit on collection when deposit done makes new deposit with
+ parent
"""
- def setUp(self):
- super().setUp()
- # Add another user
- _collection2 = DepositCollection(name='some')
- _collection2.save()
- _user = DepositClient.objects.create_user(username='user',
- password='user')
- _user.collections = [_collection2.id]
- self.collection2 = _collection2
-
- def test_access_to_another_user_collection_is_forbidden(self):
- """Access to another user collection should return a 403
-
- """
- url = reverse(COL_IRI, args=[self.collection2.name])
- response = self.client.post(url)
- self.assertEqual(response.status_code,
- status.HTTP_403_FORBIDDEN)
- self.assertRegex(response.content.decode('utf-8'),
- 'Client hal cannot access collection %s' % (
- self.collection2.name, ))
-
- def test_delete_on_col_iri_not_supported(self):
- """Delete on col iri should return a 405 response
-
- """
- url = reverse(COL_IRI, args=[self.collection.name])
- response = self.client.delete(url)
- self.assertEqual(response.status_code,
- status.HTTP_405_METHOD_NOT_ALLOWED)
- self.assertRegex(response.content.decode('utf-8'),
- 'DELETE method is not supported on this endpoint')
-
- def create_deposit_with_rejection_status(self):
- url = reverse(COL_IRI, args=[self.collection.name])
-
- data = b'some data which is clearly not a zip file'
- md5sum = hashlib.md5(data).hexdigest()
- external_id = 'some-external-id-1'
-
- # when
- response = self.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,
- HTTP_CONTENT_MD5=md5sum,
- HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip',
- HTTP_CONTENT_DISPOSITION='attachment; filename=filename0')
-
- self.assertEqual(response.status_code, status.HTTP_201_CREATED)
- response_content = parse_xml(BytesIO(response.content))
- actual_state = response_content['deposit_status']
- self.assertEqual(actual_state, DEPOSIT_STATUS_REJECTED)
-
- def test_act_on_deposit_rejected_is_not_permitted(self):
- deposit_id = self.create_deposit_with_status(DEPOSIT_STATUS_REJECTED)
-
- deposit = Deposit.objects.get(pk=deposit_id)
- assert deposit.status == DEPOSIT_STATUS_REJECTED
-
- response = self.client.post(
- reverse(EDIT_SE_IRI, args=[self.collection.name, deposit_id]),
- content_type='application/atom+xml;type=entry',
- data=self.atom_entry_data1,
- HTTP_SLUG='external-id')
-
- self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
- self.assertRegex(
- response.content.decode('utf-8'),
- "You can only act on deposit with status '%s'" % (
- DEPOSIT_STATUS_PARTIAL, ))
-
- def test_add_deposit_with_parent(self):
- # given multiple deposit already loaded
- deposit_id = self.create_deposit_with_status(
- status=DEPOSIT_STATUS_LOAD_SUCCESS,
- external_id='some-external-id')
-
- deposit1 = Deposit.objects.get(pk=deposit_id)
- self.assertIsNotNone(deposit1)
- self.assertEqual(deposit1.external_id, 'some-external-id')
- self.assertEqual(deposit1.status, DEPOSIT_STATUS_LOAD_SUCCESS)
-
- deposit_id2 = self.create_deposit_with_status(
- status=DEPOSIT_STATUS_LOAD_SUCCESS,
- external_id='some-external-id')
-
- deposit2 = Deposit.objects.get(pk=deposit_id2)
- self.assertIsNotNone(deposit2)
- self.assertEqual(deposit2.external_id, 'some-external-id')
- self.assertEqual(deposit2.status, DEPOSIT_STATUS_LOAD_SUCCESS)
-
- deposit_id3 = self.create_deposit_with_status(
- status=DEPOSIT_STATUS_LOAD_FAILURE,
- external_id='some-external-id')
-
- deposit3 = Deposit.objects.get(pk=deposit_id3)
- self.assertIsNotNone(deposit3)
- self.assertEqual(deposit3.external_id, 'some-external-id')
- self.assertEqual(deposit3.status, DEPOSIT_STATUS_LOAD_FAILURE)
-
- # when
- deposit_id3 = self.create_simple_deposit_partial(
- external_id='some-external-id')
-
- # then
- deposit4 = Deposit.objects.get(pk=deposit_id3)
-
- self.assertIsNotNone(deposit4)
- self.assertEqual(deposit4.external_id, 'some-external-id')
- self.assertEqual(deposit4.status, DEPOSIT_STATUS_PARTIAL)
- self.assertEqual(deposit4.parent, deposit2)
+ # given multiple deposit already loaded
+ deposit = completed_deposit
+ assert deposit.status == DEPOSIT_STATUS_LOAD_SUCCESS
+
+ # adding a new deposit with the same external id as a completed deposit
+ # creates the parenting chain
+ response = authenticated_client.post(
+ reverse(COL_IRI, args=[deposit_collection.name]),
+ content_type='application/atom+xml;type=entry',
+ data=atom_dataset['entry-data0'] % deposit.external_id.encode('utf-8'),
+ HTTP_SLUG=deposit.external_id
+ )
+
+ assert response.status_code == status.HTTP_201_CREATED
+ response_content = parse_xml(BytesIO(response.content))
+ deposit_id = response_content['deposit_id']
+
+ assert deposit_id != deposit.id
+
+ new_deposit = Deposit.objects.get(pk=deposit_id)
+ assert deposit.collection == new_deposit.collection
+ assert deposit.external_id == new_deposit.external_id
+
+ assert new_deposit != deposit
+ assert new_deposit.parent == deposit
diff --git a/swh/deposit/tests/conftest.py b/swh/deposit/tests/conftest.py
index bd76ea38..b816e5f5 100644
--- a/swh/deposit/tests/conftest.py
+++ b/swh/deposit/tests/conftest.py
@@ -1,259 +1,257 @@
# 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, EDIT_SE_IRI, DEPOSIT_STATUS_DEPOSITED, DEPOSIT_STATUS_REJECTED,
- DEPOSIT_STATUS_PARTIAL, DEPOSIT_STATUS_LOAD_SUCCESS
+ 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
-@pytest.fixture
-def deposit_collection(db):
+def create_deposit_collection(collection_name: str):
+ """Create a deposit collection with name collection_name
+
+ """
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
+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):
+ 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')
# 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'.
- """
- deposit = create_deposit(
- authenticated_client, deposit_collection.name, sample_archive,
- external_id='external-id-deposited')
- assert deposit.status == DEPOSIT_STATUS_DEPOSITED
+ if deposit.status != deposit_status:
+ deposit.status = deposit_status
+ deposit.save()
+ assert deposit.status == deposit_status
return deposit
-@pytest.fixture
-def rejected_deposit(sample_archive, deposit_collection, authenticated_client):
- """Returns a deposit with status 'rejected'.
+def deposit_factory(deposit_status=DEPOSIT_STATUS_DEPOSITED):
+ """Build deposit with a specific status
"""
- 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 _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
-@pytest.fixture
-def partial_deposit(sample_archive, deposit_collection, authenticated_client):
- """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
+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(
authenticated_client, deposit_collection.name, sample_archive,
- external_id='external-id-partial'
+ external_id='external-id-partial',
+ deposit_status=DEPOSIT_STATUS_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'
+ external_id='external-id-complete',
+ deposit_status=DEPOSIT_STATUS_LOAD_SUCCESS
)
- 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