Page Menu
Home
Software Heritage
Search
Configure Global Search
Log In
Files
F9312764
tests_deposit.py
No One
Temporary
Actions
Download File
Edit File
Delete File
View Transforms
Subscribe
Mute Notifications
Award Token
Flag For Later
Size
25 KB
Subscribers
None
tests_deposit.py
View Options
# Copyright (C) 2017 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.core.files.uploadedfile
import
InMemoryUploadedFile
from
django.core.urlresolvers
import
reverse
from
io
import
BytesIO
from
rest_framework
import
status
from
rest_framework.test
import
APITestCase
from
swh.deposit.models
import
Deposit
,
DepositRequest
from
swh.deposit.parsers
import
parse_xml
from
.common
import
BasicTestCase
,
WithAuthTestCase
class
DepositNoAuthCase
(
APITestCase
,
BasicTestCase
):
"""Deposit access are protected with basic authentication.
"""
def
test_post_will_fail_with_401
(
self
):
"""Without authentication, endpoint refuses access with 401 response
"""
url
=
reverse
(
'upload'
,
args
=
[
'hal'
])
data_text
=
b
'some content'
md5sum
=
hashlib
.
md5
(
data_text
)
.
hexdigest
()
external_id
=
'some-external-id-1'
# when
response
=
self
.
client
.
post
(
url
,
content_type
=
'application/zip'
,
# as zip
data
=
data_text
,
# + headers
HTTP_SLUG
=
external_id
,
HTTP_CONTENT_MD5
=
md5sum
,
HTTP_PACKAGING
=
'http://purl.org/net/sword/package/SimpleZip'
,
HTTP_IN_PROGRESS
=
'false'
,
HTTP_CONTENT_LENGTH
=
len
(
data_text
),
HTTP_CONTENT_DISPOSITION
=
'attachment; filename=filename0'
)
# then
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_401_UNAUTHORIZED
)
class
DepositTestCase
(
APITestCase
,
WithAuthTestCase
,
BasicTestCase
):
"""Try and upload one single deposit
"""
def
setUp
(
self
):
super
()
.
setUp
()
self
.
atom_entry_data0
=
b
"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom">
<title>Awesome Compiler</title>
<client>hal</client>
<id>urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a</id>
<external_identifier>%s</external_identifier>
<updated>2017-10-07T15:17:08Z</updated>
<author>some awesome author</author>
<applicationCategory>something</applicationCategory>
<name>awesome-compiler</name>
<description>This is an awesome compiler destined to
awesomely compile stuff
and other stuff</description>
<keywords>compiler,programming,language</keywords>
<dateCreated>2005-10-07T17:17:08Z</dateCreated>
<datePublished>2005-10-07T17:17:08Z</datePublished>
<releaseNotes>release note</releaseNotes>
<relatedLink>related link</relatedLink>
<sponsor></sponsor>
<programmingLanguage>Awesome</programmingLanguage>
<codeRepository>https://hoster.org/awesome-compiler</codeRepository>
<operatingSystem>GNU/Linux</operatingSystem>
<version>0.0.1</version>
<developmentStatus>running</developmentStatus>
<runtimePlatform>all</runtimePlatform>
</entry>"""
self
.
atom_entry_data1
=
b
"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom">
<client>hal</client>
<id>urn:uuid:2225c695-cfb8-4ebb-aaaa-80da344efa6a</id>
<updated>2017-10-07T15:17:08Z</updated>
<author>some awesome author</author>
<applicationCategory>something</applicationCategory>
<name>awesome-compiler</name>
<description>This is an awesome compiler destined to
awesomely compile stuff
and other stuff</description>
<keywords>compiler,programming,language</keywords>
<dateCreated>2005-10-07T17:17:08Z</dateCreated>
<datePublished>2005-10-07T17:17:08Z</datePublished>
<releaseNotes>release note</releaseNotes>
<relatedLink>related link</relatedLink>
<sponsor></sponsor>
<programmingLanguage>Awesome</programmingLanguage>
<codeRepository>https://hoster.org/awesome-compiler</codeRepository>
<operatingSystem>GNU/Linux</operatingSystem>
<version>0.0.1</version>
<developmentStatus>running</developmentStatus>
<runtimePlatform>all</runtimePlatform>
</entry>"""
self
.
atom_entry_data2
=
b
"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom">
<external_identifier>%s</external_identifier>
</entry>"""
self
.
atom_entry_data_empty_body
=
b
"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom"></entry>"""
self
.
atom_entry_data3
=
b
"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom">
<something>something</something>
</entry>"""
self
.
data_atom_entry_ok
=
b
"""<?xml version="1.0"?>
<entry xmlns="http://www.w3.org/2005/Atom"
xmlns:dcterms="http://purl.org/dc/terms/">
<title>Title</title>
<id>urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a</id>
<updated>2005-10-07T17:17:08Z</updated>
<author><name>Contributor</name></author>
<summary type="text">The abstract</summary>
<!-- some embedded metadata -->
<dcterms:abstract>The abstract</dcterms:abstract>
<dcterms:accessRights>Access Rights</dcterms:accessRights>
<dcterms:alternative>Alternative Title</dcterms:alternative>
<dcterms:available>Date Available</dcterms:available>
<dcterms:bibliographicCitation>Bibliographic Citation</dcterms:bibliographicCitation> # noqa
<dcterms:contributor>Contributor</dcterms:contributor>
<dcterms:description>Description</dcterms:description>
<dcterms:hasPart>Has Part</dcterms:hasPart>
<dcterms:hasVersion>Has Version</dcterms:hasVersion>
<dcterms:identifier>Identifier</dcterms:identifier>
<dcterms:isPartOf>Is Part Of</dcterms:isPartOf>
<dcterms:publisher>Publisher</dcterms:publisher>
<dcterms:references>References</dcterms:references>
<dcterms:rightsHolder>Rights Holder</dcterms:rightsHolder>
<dcterms:source>Source</dcterms:source>
<dcterms:title>Title</dcterms:title>
<dcterms:type>Type</dcterms:type>
</entry>"""
def
test_post_deposit_binary_upload_final
(
self
):
"""Binary upload should be accepted
"""
# given
url
=
reverse
(
'upload'
,
args
=
[
'hal'
])
data_text
=
b
'some content'
md5sum
=
hashlib
.
md5
(
data_text
)
.
hexdigest
()
id
=
hashlib
.
sha1
(
data_text
)
.
hexdigest
()
external_id
=
'some-external-id-1'
# when
response
=
self
.
client
.
post
(
url
,
content_type
=
'application/zip'
,
# as zip
data
=
data_text
,
# + headers
HTTP_SLUG
=
external_id
,
HTTP_CONTENT_MD5
=
md5sum
,
HTTP_PACKAGING
=
'http://purl.org/net/sword/package/SimpleZip'
,
HTTP_IN_PROGRESS
=
'false'
,
HTTP_CONTENT_LENGTH
=
len
(
data_text
),
HTTP_CONTENT_DISPOSITION
=
'attachment; filename=filename0'
)
# then
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_201_CREATED
)
deposit
=
Deposit
.
objects
.
get
(
external_id
=
external_id
)
self
.
assertIsNotNone
(
deposit
)
self
.
assertEqual
(
deposit
.
status
,
'ready'
)
self
.
assertEqual
(
deposit
.
external_id
,
external_id
)
self
.
assertEqual
(
deposit
.
type
,
self
.
type
)
self
.
assertEqual
(
deposit
.
client
,
self
.
user
)
self
.
assertIsNone
(
deposit
.
swh_id
)
deposit_request
=
DepositRequest
.
objects
.
get
(
deposit
=
deposit
)
self
.
assertIsNotNone
(
deposit_request
)
self
.
assertEquals
(
deposit_request
.
deposit
,
deposit
)
self
.
assertEquals
(
deposit_request
.
metadata
,
{
'archive'
:
{
'id'
:
id
,
'name'
:
'filename0'
,
},
})
response_content
=
parse_xml
(
BytesIO
(
response
.
content
))
self
.
assertEqual
(
response_content
[
'{http://www.w3.org/2005/Atom}deposit_archive'
],
'filename0'
)
self
.
assertEqual
(
response_content
[
'{http://www.w3.org/2005/Atom}deposit_id'
],
deposit
.
id
)
def
test_post_deposit_binary_upload_2_steps
(
self
):
"""Binary upload should be accepted
"""
# given
url
=
reverse
(
'upload'
,
args
=
[
'hal'
])
external_id
=
'some-external-id-1'
# 1st archive to upload
data_text0
=
b
'some other content'
md5sum0
=
hashlib
.
md5
(
data_text0
)
.
hexdigest
()
id0
=
hashlib
.
sha1
(
data_text0
)
.
hexdigest
()
# when
response
=
self
.
client
.
post
(
url
,
content_type
=
'application/zip'
,
# as zip
data
=
data_text0
,
# + headers
HTTP_SLUG
=
external_id
,
HTTP_CONTENT_MD5
=
md5sum0
,
HTTP_PACKAGING
=
'http://purl.org/net/sword/package/SimpleZip'
,
HTTP_IN_PROGRESS
=
'true'
,
HTTP_CONTENT_LENGTH
=
len
(
data_text0
),
HTTP_CONTENT_DISPOSITION
=
'attachment; filename=filename0'
)
# then
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_201_CREATED
)
deposit
=
Deposit
.
objects
.
get
(
external_id
=
external_id
)
self
.
assertIsNotNone
(
deposit
)
self
.
assertEqual
(
deposit
.
status
,
'partial'
)
self
.
assertEqual
(
deposit
.
external_id
,
external_id
)
self
.
assertEqual
(
deposit
.
type
,
self
.
type
)
self
.
assertEqual
(
deposit
.
client
,
self
.
user
)
self
.
assertIsNone
(
deposit
.
swh_id
)
deposit_request
=
DepositRequest
.
objects
.
get
(
deposit
=
deposit
)
self
.
assertIsNotNone
(
deposit_request
)
self
.
assertEquals
(
deposit_request
.
deposit
,
deposit
)
self
.
assertEquals
(
deposit_request
.
metadata
,
{
'archive'
:
{
'id'
:
id0
,
'name'
:
'filename0'
,
},
})
# 2nd archive to upload
data_text
=
b
'second archive uploaded'
md5sum1
=
hashlib
.
md5
(
data_text
)
.
hexdigest
()
id1
=
hashlib
.
sha1
(
data_text
)
.
hexdigest
()
response
=
self
.
client
.
post
(
url
,
content_type
=
'application/zip'
,
# as zip
data
=
data_text
,
# + headers
HTTP_SLUG
=
external_id
,
HTTP_CONTENT_MD5
=
md5sum1
,
HTTP_PACKAGING
=
'http://purl.org/net/sword/package/SimpleZip'
,
HTTP_IN_PROGRESS
=
'false'
,
HTTP_CONTENT_LENGTH
=
len
(
data_text
),
HTTP_CONTENT_DISPOSITION
=
'attachment; filename=filename1'
)
deposit
=
Deposit
.
objects
.
get
(
external_id
=
external_id
)
self
.
assertIsNotNone
(
deposit
)
self
.
assertEqual
(
deposit
.
status
,
'ready'
)
self
.
assertEqual
(
deposit
.
external_id
,
external_id
)
self
.
assertEqual
(
deposit
.
type
,
self
.
type
)
self
.
assertEqual
(
deposit
.
client
,
self
.
user
)
self
.
assertIsNone
(
deposit
.
swh_id
)
deposit_requests
=
list
(
DepositRequest
.
objects
.
filter
(
deposit
=
deposit
))
self
.
assertIsNotNone
(
deposit_requests
)
self
.
assertEquals
(
len
(
deposit_requests
),
2
)
self
.
assertEquals
(
deposit_requests
[
0
]
.
deposit
,
deposit
)
self
.
assertEquals
(
deposit_requests
[
0
]
.
metadata
,
{
'archive'
:
{
'id'
:
id0
,
'name'
:
'filename0'
,
},
})
self
.
assertEquals
(
deposit_requests
[
1
]
.
deposit
,
deposit
)
self
.
assertEquals
(
deposit_requests
[
1
]
.
metadata
,
{
'archive'
:
{
'id'
:
id1
,
'name'
:
'filename1'
,
},
})
def
test_post_deposit_binary_upload_only_supports_zip
(
self
):
"""Binary upload only supports application/zip (for now)...
"""
# given
url
=
reverse
(
'upload'
,
args
=
[
'hal'
])
data_text
=
b
'some content'
md5sum
=
hashlib
.
md5
(
data_text
)
.
hexdigest
()
external_id
=
'some-external-id-1'
# when
response
=
self
.
client
.
post
(
url
,
content_type
=
'application/octet-stream'
,
data
=
data_text
,
# + headers
HTTP_SLUG
=
external_id
,
HTTP_CONTENT_MD5
=
md5sum
,
HTTP_PACKAGING
=
'http://purl.org/net/sword/package/SimpleZip'
,
HTTP_IN_PROGRESS
=
'false'
,
HTTP_CONTENT_LENGTH
=
len
(
data_text
),
HTTP_CONTENT_DISPOSITION
=
'attachment; filename=filename0'
)
# then
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_415_UNSUPPORTED_MEDIA_TYPE
)
try
:
Deposit
.
objects
.
get
(
external_id
=
external_id
)
except
Deposit
.
DoesNotExist
:
pass
def
test_post_deposit_binary_fails_if_unsupported_packaging_header
(
self
):
"""Binary upload must have content_disposition header provided...
"""
# given
url
=
reverse
(
'upload'
,
args
=
[
'hal'
])
data_text
=
b
'some content'
md5sum
=
hashlib
.
md5
(
data_text
)
.
hexdigest
()
external_id
=
'some-external-id'
# when
response
=
self
.
client
.
post
(
url
,
content_type
=
'application/zip'
,
data
=
data_text
,
# + headers
HTTP_SLUG
=
external_id
,
HTTP_CONTENT_MD5
=
md5sum
,
HTTP_PACKAGING
=
'something-unsupported'
,
HTTP_CONTENT_LENGTH
=
len
(
data_text
),
HTTP_CONTENT_DISPOSITION
=
'attachment; filename=filename0'
)
# then
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_400_BAD_REQUEST
)
try
:
Deposit
.
objects
.
get
(
external_id
=
external_id
)
except
Deposit
.
DoesNotExist
:
pass
def
test_post_deposit_binary_upload_fail_if_no_content_disposition_header
(
self
):
"""Binary upload must have content_disposition header provided...
"""
# given
url
=
reverse
(
'upload'
,
args
=
[
'hal'
])
data_text
=
b
'some content'
md5sum
=
hashlib
.
md5
(
data_text
)
.
hexdigest
()
external_id
=
'some-external-id'
# when
response
=
self
.
client
.
post
(
url
,
content_type
=
'application/zip'
,
data
=
data_text
,
# + headers
HTTP_SLUG
=
external_id
,
HTTP_CONTENT_MD5
=
md5sum
,
HTTP_PACKAGING
=
'http://purl.org/net/sword/package/SimpleZip'
,
HTTP_IN_PROGRESS
=
'false'
,
HTTP_CONTENT_LENGTH
=
len
(
data_text
))
# then
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_400_BAD_REQUEST
)
try
:
Deposit
.
objects
.
get
(
external_id
=
external_id
)
except
Deposit
.
DoesNotExist
:
pass
def
test_post_deposit_mediation_not_supported
(
self
):
"""Binary upload only supports application/zip (for now)...
"""
# given
url
=
reverse
(
'upload'
,
args
=
[
'hal'
])
data_text
=
b
'some content'
md5sum
=
hashlib
.
md5
(
data_text
)
.
hexdigest
()
external_id
=
'some-external-id-1'
# when
response
=
self
.
client
.
post
(
url
,
content_type
=
'application/zip'
,
data
=
data_text
,
# + headers
HTTP_SLUG
=
external_id
,
HTTP_CONTENT_MD5
=
md5sum
,
HTTP_PACKAGING
=
'http://purl.org/net/sword/package/SimpleZip'
,
HTTP_IN_PROGRESS
=
'false'
,
HTTP_CONTENT_LENGTH
=
len
(
data_text
),
HTTP_ON_BEHALF_OF
=
'someone'
,
HTTP_CONTENT_DISPOSITION
=
'attachment; filename=filename0'
)
# then
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_412_PRECONDITION_FAILED
)
try
:
Deposit
.
objects
.
get
(
external_id
=
external_id
)
except
Deposit
.
DoesNotExist
:
pass
# FIXME: Test this scenario (need a way to override the default
# size limit in test scenario)
# def test_post_deposit_binary_upload_fail_if_upload_size_limit_exceeded(
# self):
# """Binary upload must not exceed the limit set up...
# """
# # given
# url = reverse('upload', args=['hal'])
# data_text = b'some content'
# md5sum = hashlib.md5(data_text).hexdigest()
# external_id = 'some-external-id'
# # when
# response = self.client.post(
# url,
# content_type='application/zip',
# data=data_text,
# # + headers
# HTTP_SLUG=external_id,
# HTTP_CONTENT_MD5=md5sum,
# HTTP_PACKAGING='http://purl.org/net/sword/package/SimpleZip',
# HTTP_IN_PROGRESS='false',
# CONTENT_LENGTH=len(data_text),
# HTTP_CONTENT_DISPOSITION='attachment; filename=filename0')
# # then
# self.assertEqual(response.status_code,
# status.HTTP_403_FORBIDDEN)
# try:
# Deposit.objects.get(external_id=external_id)
# except Deposit.DoesNotExist:
# pass
def
test_post_deposit_multipart
(
self
):
"""Test one deposit upload."""
# given
url
=
reverse
(
'upload'
,
args
=
[
'hal'
])
# from django.core.files import uploadedfile
data_atom_entry
=
self
.
data_atom_entry_ok
archive_content
=
b
'some content representing archive'
archive
=
InMemoryUploadedFile
(
BytesIO
(
archive_content
),
field_name
=
'archive0'
,
name
=
'archive0'
,
content_type
=
'application/zip'
,
size
=
len
(
archive_content
),
charset
=
None
)
atom_entry
=
InMemoryUploadedFile
(
BytesIO
(
data_atom_entry
),
field_name
=
'atom0'
,
name
=
'atom0'
,
content_type
=
'application/atom+xml; charset="utf-8"'
,
size
=
len
(
data_atom_entry
),
charset
=
'utf-8'
)
external_id
=
'external-id'
id1
=
hashlib
.
sha1
(
archive_content
)
.
hexdigest
()
# when
response
=
self
.
client
.
post
(
url
,
format
=
'multipart'
,
data
=
{
'archive'
:
archive
,
'atom_entry'
:
atom_entry
,
},
# + headers
HTTP_IN_PROGRESS
=
'false'
,
HTTP_SLUG
=
external_id
)
# then
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_201_CREATED
)
deposit
=
Deposit
.
objects
.
get
(
external_id
=
external_id
)
self
.
assertIsNotNone
(
deposit
)
self
.
assertEqual
(
deposit
.
status
,
'ready'
)
self
.
assertEqual
(
deposit
.
external_id
,
external_id
)
self
.
assertEqual
(
deposit
.
type
,
self
.
type
)
self
.
assertEqual
(
deposit
.
client
,
self
.
user
)
self
.
assertIsNone
(
deposit
.
swh_id
)
deposit_request
=
DepositRequest
.
objects
.
get
(
deposit
=
deposit
)
self
.
assertIsNotNone
(
deposit_request
)
self
.
assertEquals
(
deposit_request
.
deposit
,
deposit
)
self
.
assertEquals
(
deposit_request
.
metadata
[
'archive'
],
{
'id'
:
id1
,
'name'
:
'archive0'
,
})
# then
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_201_CREATED
)
deposit
=
Deposit
.
objects
.
get
(
external_id
=
external_id
)
self
.
assertIsNotNone
(
deposit
)
self
.
assertEqual
(
deposit
.
status
,
'ready'
)
self
.
assertEqual
(
deposit
.
external_id
,
external_id
)
self
.
assertEqual
(
deposit
.
type
,
self
.
type
)
self
.
assertEqual
(
deposit
.
client
,
self
.
user
)
self
.
assertIsNone
(
deposit
.
swh_id
)
deposit_request
=
DepositRequest
.
objects
.
get
(
deposit
=
deposit
)
self
.
assertIsNotNone
(
deposit_request
)
self
.
assertEquals
(
deposit_request
.
deposit
,
deposit
)
self
.
assertEquals
(
deposit_request
.
metadata
[
'archive'
],
{
'id'
:
id1
,
'name'
:
'archive0'
,
})
def
test_post_deposit_multipart_only_archive_and_atom_entry
(
self
):
"""Multipart deposit only accepts one archive and one atom+xml"""
# given
url
=
reverse
(
'upload'
,
args
=
[
'hal'
])
# from django.core.files import uploadedfile
archive_content
=
b
'some content representing archive'
archive
=
InMemoryUploadedFile
(
BytesIO
(
archive_content
),
field_name
=
'archive0'
,
name
=
'archive0'
,
content_type
=
'application/zip'
,
size
=
len
(
archive_content
),
charset
=
None
)
other_archive_content
=
b
"some-other-content"
other_archive
=
InMemoryUploadedFile
(
BytesIO
(
other_archive_content
),
field_name
=
'atom0'
,
name
=
'atom0'
,
content_type
=
'application/zip'
,
size
=
len
(
other_archive_content
),
charset
=
'utf-8'
)
# when
response
=
self
.
client
.
post
(
url
,
format
=
'multipart'
,
data
=
{
'archive'
:
archive
,
'atom_entry'
:
other_archive
,
},
# + headers
HTTP_IN_PROGRESS
=
'false'
,
HTTP_SLUG
=
'external-id'
)
# then
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_415_UNSUPPORTED_MEDIA_TYPE
)
# when
archive
.
seek
(
0
)
response
=
self
.
client
.
post
(
url
,
format
=
'multipart'
,
data
=
{
'archive'
:
archive
,
},
# + headers
HTTP_IN_PROGRESS
=
'false'
,
HTTP_SLUG
=
'external-id'
)
# then
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_415_UNSUPPORTED_MEDIA_TYPE
)
def
test_post_deposit_atom_empty_body_request
(
self
):
response
=
self
.
client
.
post
(
reverse
(
'upload'
,
args
=
[
'hal'
]),
content_type
=
'application/atom+xml;type=entry'
,
data
=
self
.
atom_entry_data_empty_body
)
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_400_BAD_REQUEST
)
def
test_post_deposit_atom_unknown_no_external_id
(
self
):
response
=
self
.
client
.
post
(
reverse
(
'upload'
,
args
=
[
'hal'
]),
content_type
=
'application/atom+xml;type=entry'
,
data
=
self
.
atom_entry_data3
)
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_400_BAD_REQUEST
)
def
test_post_deposit_atom_unknown_client
(
self
):
response
=
self
.
client
.
post
(
reverse
(
'upload'
,
args
=
[
'unknown-one'
]),
content_type
=
'application/atom+xml;type=entry'
,
data
=
self
.
atom_entry_data3
,
HTTP_SLUG
=
'something'
)
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_400_BAD_REQUEST
)
def
test_post_deposit_atom_entry_initial
(
self
):
"""One deposit upload as atom entry
"""
# given
external_id
=
'urn:uuid:1225c695-cfb8-4ebb-aaaa-80da344efa6a'
try
:
deposit
=
Deposit
.
objects
.
get
(
external_id
=
external_id
)
except
Deposit
.
DoesNotExist
:
assert
True
atom_entry_data
=
self
.
atom_entry_data0
%
external_id
.
encode
(
'utf-8'
)
# when
response
=
self
.
client
.
post
(
reverse
(
'upload'
,
args
=
[
'hal'
]),
content_type
=
'application/atom+xml;type=entry'
,
data
=
atom_entry_data
,
HTTP_IN_PROGRESS
=
'false'
)
# then
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_201_CREATED
)
deposit
=
Deposit
.
objects
.
get
(
external_id
=
external_id
)
self
.
assertIsNotNone
(
deposit
)
self
.
assertEqual
(
deposit
.
type
,
self
.
type
)
self
.
assertEqual
(
deposit
.
external_id
,
external_id
)
self
.
assertEqual
(
deposit
.
status
,
'ready'
)
self
.
assertEqual
(
deposit
.
client
,
self
.
user
)
# one associated request to a deposit
deposit_request
=
DepositRequest
.
objects
.
get
(
deposit
=
deposit
)
actual_metadata
=
deposit_request
.
metadata
self
.
assertIsNotNone
(
actual_metadata
)
self
.
assertIsNone
(
actual_metadata
.
get
(
'archive'
))
def
test_post_deposit_atom_entry_multiple_step
(
self
):
"""Test one deposit upload."""
# given
external_id
=
'urn:uuid:2225c695-cfb8-4ebb-aaaa-80da344efa6a'
try
:
deposit
=
Deposit
.
objects
.
get
(
external_id
=
external_id
)
except
Deposit
.
DoesNotExist
:
assert
True
# when
response
=
self
.
client
.
post
(
reverse
(
'upload'
,
args
=
[
'hal'
]),
content_type
=
'application/atom+xml;type=entry'
,
data
=
self
.
atom_entry_data1
,
HTTP_IN_PROGRESS
=
'True'
,
HTTP_SLUG
=
external_id
)
# then
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_201_CREATED
)
deposit
=
Deposit
.
objects
.
get
(
external_id
=
external_id
)
self
.
assertIsNotNone
(
deposit
)
self
.
assertEqual
(
deposit
.
type
,
self
.
type
)
self
.
assertEqual
(
deposit
.
external_id
,
external_id
)
self
.
assertEqual
(
deposit
.
status
,
'partial'
)
self
.
assertEqual
(
deposit
.
client
,
self
.
user
)
# one associated request to a deposit
deposit_requests
=
DepositRequest
.
objects
.
filter
(
deposit
=
deposit
)
self
.
assertEqual
(
len
(
deposit_requests
),
1
)
atom_entry_data
=
self
.
atom_entry_data2
%
external_id
.
encode
(
'utf-8'
)
# when
response
=
self
.
client
.
post
(
reverse
(
'upload'
,
args
=
[
'hal'
]),
content_type
=
'application/atom+xml;type=entry'
,
data
=
atom_entry_data
,
HTTP_IN_PROGRESS
=
'False'
,
HTTP_SLUG
=
external_id
)
# then
self
.
assertEqual
(
response
.
status_code
,
status
.
HTTP_201_CREATED
)
deposit
=
Deposit
.
objects
.
get
(
external_id
=
external_id
)
self
.
assertIsNotNone
(
deposit
)
self
.
assertEqual
(
deposit
.
type
,
self
.
type
)
self
.
assertEqual
(
deposit
.
external_id
,
external_id
)
self
.
assertEqual
(
deposit
.
status
,
'ready'
)
self
.
assertEqual
(
deposit
.
client
,
self
.
user
)
self
.
assertEqual
(
len
(
Deposit
.
objects
.
all
()),
1
)
# now 2 associated requests to a same deposit
deposit_requests
=
DepositRequest
.
objects
.
filter
(
deposit
=
deposit
)
self
.
assertEqual
(
len
(
deposit_requests
),
2
)
for
deposit_request
in
deposit_requests
:
actual_metadata
=
deposit_request
.
metadata
self
.
assertIsNotNone
(
actual_metadata
)
self
.
assertIsNone
(
actual_metadata
.
get
(
'archive'
))
File Metadata
Details
Attached
Mime Type
text/x-python
Expires
Thu, Jul 3, 11:08 AM (1 w, 2 d ago)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
3452099
Attached To
rDDEP Push deposit
Event Timeline
Log In to Comment