Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/tests/common.py
# Copyright (C) 2017-2019 The Software Heritage developers | # Copyright (C) 2017-2019 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
import hashlib | import hashlib | ||||
from io import BytesIO | |||||
import os | import os | ||||
import re | import re | ||||
import tarfile | import tarfile | ||||
import tempfile | import tempfile | ||||
from django.core.files.uploadedfile import InMemoryUploadedFile | |||||
from swh.core import tarball | from swh.core import tarball | ||||
def compute_info(archive_path): | def compute_info(archive_path): | ||||
"""Given a path, compute information on path. | """Given a path, compute information on path. | ||||
""" | """ | ||||
with open(archive_path, "rb") as f: | with open(archive_path, "rb") as f: | ||||
▲ Show 20 Lines • Show All 150 Lines • ▼ Show 20 Lines | return authenticated_client.post( | ||||
url, content_type="application/atom+xml;type=entry", data=data, **kwargs | url, content_type="application/atom+xml;type=entry", data=data, **kwargs | ||||
) | ) | ||||
def put_atom(authenticated_client, url, data, **kwargs): | def put_atom(authenticated_client, url, data, **kwargs): | ||||
return authenticated_client.put( | return authenticated_client.put( | ||||
url, content_type="application/atom+xml;type=entry", data=data, **kwargs | url, content_type="application/atom+xml;type=entry", data=data, **kwargs | ||||
) | ) | ||||
def _post_or_put_multipart(f, url, archive, atom_entry, **kwargs): | |||||
archive = InMemoryUploadedFile( | |||||
BytesIO(archive["data"]), | |||||
field_name=archive["name"], | |||||
name=archive["name"], | |||||
content_type="application/x-tar", | |||||
size=archive["length"], | |||||
charset=None, | |||||
) | |||||
atom_entry = InMemoryUploadedFile( | |||||
BytesIO(atom_entry.encode("utf-8")), | |||||
field_name="atom0", | |||||
name="atom0", | |||||
content_type='application/atom+xml; charset="utf-8"', | |||||
size=len(atom_entry), | |||||
charset="utf-8", | |||||
) | |||||
return f( | |||||
url, | |||||
format="multipart", | |||||
data={"archive": archive, "atom_entry": atom_entry,}, | |||||
**kwargs, | |||||
) | |||||
def post_multipart(authenticated_client, *args, **kwargs): | |||||
return _post_or_put_multipart(authenticated_client.post, *args, **kwargs) | |||||
def put_multipart(authenticated_client, *args, **kwargs): | |||||
return _post_or_put_multipart(authenticated_client.put, *args, **kwargs) |