Changeset View
Changeset View
Standalone 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): | ||||
ardumont: Here is the code to compress an archive according to latest remark. | |||||
Not Done Inline Actionslatest remark is D380#inline-1900 ;) ardumont: latest remark is D380#inline-1900 ;) | |||||
"""Define the test client and other test variables.""" | """Define the test client and other test variables.""" | ||||
super().setUp() | super().setUp() | ||||
self.root_path = '/tmp/swh-deposit/test/build-zip/' | self.root_path = '/tmp/swh-deposit/test/build-zip/' | ||||
os.makedirs(self.root_path, exist_ok=True) | os.makedirs(self.root_path, exist_ok=True) | ||||
self.archive = create_arborescence_zip( | self.archive = create_arborescence_zip( | ||||
self.root_path, 'archive1', 'file1', b'some content in file') | self.root_path, 'archive1', 'file1', b'some content in file') | ||||
▲ Show 20 Lines • Show All 52 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 |
Here is the code to compress an archive according to latest remark.