From 2de3ee30bc07507250defaeaf1dd130ba350d67b Mon Sep 17 00:00:00 2001
From: "Antoine R. Dumont (@ardumont)" <antoine.romain.dumont@gmail.com>
Date: Sun, 20 Oct 2019 17:31:19 +0200
Subject: [PATCH] test_deposit_update: Migrate to pytest
---
swh/deposit/tests/api/test_deposit_update.py | 646 ++++++++++---------
swh/deposit/tests/conftest.py | 58 +-
2 files changed, 388 insertions(+), 316 deletions(-)
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
@@ -5,329 +5,379 @@
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"""<?xml version="1.0"?>
-<entry xmlns="http://www.w3.org/2005/Atom">
- <foobar>bar</foobar>
-</entry>"""
-
- self.atom_entry_data1 = b"""<?xml version="1.0"?>
-<entry xmlns="http://www.w3.org/2005/Atom">
- <foobar>bar</foobar>
-</entry>"""
-
- 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
@@ -11,8 +11,8 @@ 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 (
@@ -86,7 +86,8 @@ def create_deposit_collection(collection_name: str):
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)
@@ -171,7 +172,8 @@ def create_deposit(
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
@@ -185,6 +187,38 @@ def create_deposit(
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
@@ -215,25 +249,13 @@ def partial_deposit_with_metadata(
"""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):
--
GitLab