Differential D381 Diff 1200 swh/deposit/tests/common.py

Changeset View

Standalone View

swh/deposit/tests/common.py

# Copyright (C) 2017-2018 The Software Heritage developers		# Copyright (C) 2017-2018 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 base64		import base64
import hashlib		import hashlib
import os		import os
import shutil		import shutil
		import tarfile
import tempfile		import tempfile

from django.core.urlresolvers import reverse		from django.core.urlresolvers import reverse
from django.test import TestCase		from django.test import TestCase
from io import BytesIO		from io import BytesIO
from nose.plugins.attrib import attr		from nose.plugins.attrib import attr
from rest_framework import status		from rest_framework import status

from swh.deposit.config import (COL_IRI, EM_IRI, EDIT_SE_IRI,		from swh.deposit.config import (COL_IRI, EM_IRI, EDIT_SE_IRI,
DEPOSIT_STATUS_PARTIAL,		DEPOSIT_STATUS_PARTIAL,
DEPOSIT_STATUS_VERIFIED,		DEPOSIT_STATUS_VERIFIED,
DEPOSIT_STATUS_REJECTED,		DEPOSIT_STATUS_REJECTED,
DEPOSIT_STATUS_DEPOSITED)		DEPOSIT_STATUS_DEPOSITED)
from swh.deposit.models import DepositClient, DepositCollection, Deposit		from swh.deposit.models import DepositClient, DepositCollection, Deposit
from swh.deposit.models import DepositRequest		from swh.deposit.models import DepositRequest
from swh.deposit.models import DepositRequestType		from swh.deposit.models import DepositRequestType
from swh.deposit.parsers import parse_xml		from swh.deposit.parsers import parse_xml
from swh.deposit.settings.testing import MEDIA_ROOT		from swh.deposit.settings.testing import MEDIA_ROOT
from swh.core import tarball		from swh.core import tarball


		def compute_info(archive_path):
		"""Given a path, compute information on path.

		"""
		with open(archive_path, 'rb') as f:
		length = 0
		sha1sum = hashlib.sha1()
		md5sum = hashlib.md5()
		data = b''
		for chunk in f:
		sha1sum.update(chunk)
		md5sum.update(chunk)
		length += len(chunk)
		data += chunk

		return {
		'dir': os.path.dirname(archive_path),
		'name': os.path.basename(archive_path),
		'path': archive_path,
		'length': length,
		'sha1sum': sha1sum.hexdigest(),
		'md5sum': md5sum.hexdigest(),
		'data': data
		}


def create_arborescence_zip(root_path, archive_name, filename, content,		def create_arborescence_zip(root_path, archive_name, filename, content,
up_to_size=None):		up_to_size=None):
"""Build an archive named archive_name in the root_path.		"""Build an archive named archive_name in the root_path.
This archive contains one file named filename with the content content.		This archive contains one file named filename with the content content.

Returns:		Returns:
dict with the keys:		dict with the keys:
- dir: the directory of that archive		- dir: the directory of that archive
Show All 17 Lines	with open(filepath, 'wb') as f:
if up_to_size: # fill with blank content up to a given size		if up_to_size: # fill with blank content up to a given size
count += _length		count += _length
while count < up_to_size:		while count < up_to_size:
f.write(b'0'*batch_size)		f.write(b'0'*batch_size)
count += batch_size		count += batch_size

zip_path = dir_path + '.zip'		zip_path = dir_path + '.zip'
zip_path = tarball.compress(zip_path, 'zip', dir_path)		zip_path = tarball.compress(zip_path, 'zip', dir_path)
		return compute_info(zip_path)

with open(zip_path, 'rb') as f:
length = 0
sha1sum = hashlib.sha1()
md5sum = hashlib.md5()
data = b''
for chunk in f:
sha1sum.update(chunk)
md5sum.update(chunk)
length += len(chunk)
data += chunk

return {		def create_archive_with_archive(root_path, name, archive):
'dir': archive_path_dir,		"""Create an archive holding another.
'name': archive_name,
'data': data,		"""
'path': zip_path,		invalid_archive_path = os.path.join(root_path, name)
'sha1sum': sha1sum.hexdigest(),		with tarfile.open(invalid_archive_path, 'w:gz') as _archive:
'md5sum': md5sum.hexdigest(),		_archive.add(archive['path'], arcname=archive['name'])
'length': length,		return compute_info(invalid_archive_path)
}


@attr('fs')		@attr('fs')
class FileSystemCreationRoutine(TestCase):		class FileSystemCreationRoutine(TestCase):
"""Mixin intended for tests needed to tamper with archives.		"""Mixin intended for tests needed to tamper with archives.

"""		"""
def setUp(self):		def setUp(self):
▲ Show 20 Lines • Show All 60 Lines • ▼ Show 20 Lines	def create_complex_binary_deposit(self, status_partial=False):
HTTP_CONTENT_DISPOSITION='attachment; filename=filename1.zip')		HTTP_CONTENT_DISPOSITION='attachment; filename=filename1.zip')

# then		# then
assert response.status_code == status.HTTP_201_CREATED		assert response.status_code == status.HTTP_201_CREATED
response_content = parse_xml(BytesIO(response.content))		response_content = parse_xml(BytesIO(response.content))
deposit_id = int(response_content['deposit_id'])		deposit_id = int(response_content['deposit_id'])
return deposit_id		return deposit_id

		def create_deposit_archive_with_archive(self):
		invalid_archive = create_archive_with_archive(
		self.root_path, 'invalid.tar.gz', self.archive)

		response = self.client.post(
		reverse(COL_IRI, args=[self.collection.name]),
		content_type='application/x-tar',
		data=invalid_archive['data'],
		CONTENT_LENGTH=invalid_archive['length'],
		HTTP_MD5SUM=invalid_archive['md5sum'],
		HTTP_SLUG='external-id',
		HTTP_IN_PROGRESS=False,
		HTTP_CONTENT_DISPOSITION='attachment; filename=%s' % (
		invalid_archive['name'], ))

		# then
		self.assertEqual(response.status_code, status.HTTP_201_CREATED)
		response_content = parse_xml(BytesIO(response.content))
		_status = response_content['deposit_status']
		self.assertEqual(_status, DEPOSIT_STATUS_DEPOSITED)
		deposit_id = int(response_content['deposit_id'])
		return deposit_id

def update_binary_deposit(self, deposit_id, status_partial=False):		def update_binary_deposit(self, deposit_id, status_partial=False):
# update existing deposit with atom entry metadata		# update existing deposit with atom entry metadata
response = self.client.post(		response = self.client.post(
reverse(EDIT_SE_IRI, args=[self.collection.name, deposit_id]),		reverse(EDIT_SE_IRI, args=[self.collection.name, deposit_id]),
content_type='application/atom+xml;type=entry',		content_type='application/atom+xml;type=entry',
data=self.codemeta_entry_data1,		data=self.codemeta_entry_data1,
HTTP_SLUG='external-id',		HTTP_SLUG='external-id',
HTTP_IN_PROGRESS=status_partial)		HTTP_IN_PROGRESS=status_partial)
▲ Show 20 Lines • Show All 310 Lines • Show Last 20 Lines