Skip to content
Snippets Groups Projects
Verified Commit 2de3ee30 authored by Antoine R. Dumont's avatar Antoine R. Dumont
Browse files

test_deposit_update: Migrate to pytest

parent dd68e9ad
No related branches found
No related tags found
1 merge request!40Migrate most deposit tests to pytest
......@@ -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']
......@@ -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):
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment