Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/tests/api/test_deposit_read_archive.py
| # Copyright (C) 2017 The Software Heritage developers | # Copyright (C) 2017 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 | ||||
| import os | import os | ||||
| from django.core.urlresolvers import reverse | from django.core.urlresolvers import reverse | ||||
| from nose.tools import istest | |||||
| from nose.plugins.attrib import attr | from nose.plugins.attrib import attr | ||||
| from rest_framework import status | from rest_framework import status | ||||
| from rest_framework.test import APITestCase | from rest_framework.test import APITestCase | ||||
| from swh.core import tarball | from swh.core import tarball | ||||
| from swh.deposit.config import PRIVATE_GET_RAW_CONTENT | from swh.deposit.config import PRIVATE_GET_RAW_CONTENT | ||||
| from swh.deposit.tests import TEST_CONFIG | from swh.deposit.tests import TEST_CONFIG | ||||
| from ..common import BasicTestCase, WithAuthTestCase, CommonCreationRoutine | from ..common import BasicTestCase, WithAuthTestCase, CommonCreationRoutine | ||||
| from ..common import FileSystemCreationRoutine, create_arborescence_archive | from ..common import FileSystemCreationRoutine, create_arborescence_archive | ||||
| @attr('fs') | @attr('fs') | ||||
| class DepositReadArchivesTest(APITestCase, WithAuthTestCase, | class DepositReadArchivesTest(APITestCase, WithAuthTestCase, | ||||
| BasicTestCase, CommonCreationRoutine, | BasicTestCase, CommonCreationRoutine, | ||||
| FileSystemCreationRoutine): | FileSystemCreationRoutine): | ||||
| def setUp(self): | def setUp(self): | ||||
| super().setUp() | super().setUp() | ||||
| self.archive2 = create_arborescence_archive( | self.archive2 = create_arborescence_archive( | ||||
| self.root_path, 'archive2', 'file2', b'some other content in file') | self.root_path, 'archive2', 'file2', b'some other content in file') | ||||
| self.workdir = os.path.join(self.root_path, 'workdir') | self.workdir = os.path.join(self.root_path, 'workdir') | ||||
| @istest | def test_access_to_existing_deposit_with_one_archive(self): | ||||
| def access_to_existing_deposit_with_one_archive(self): | |||||
| """Access to deposit should stream a 200 response with its raw content | """Access to deposit should stream a 200 response with its raw content | ||||
| """ | """ | ||||
| deposit_id = self.create_simple_binary_deposit() | deposit_id = self.create_simple_binary_deposit() | ||||
| url = reverse(PRIVATE_GET_RAW_CONTENT, | url = reverse(PRIVATE_GET_RAW_CONTENT, | ||||
| args=[self.collection.name, deposit_id]) | args=[self.collection.name, deposit_id]) | ||||
| Show All 22 Lines | def _check_tarball_consistency(self, actual_sha1): | ||||
| tarball.compress(new_path, 'zip', self.workdir) | tarball.compress(new_path, 'zip', self.workdir) | ||||
| with open(new_path, 'rb') as f: | with open(new_path, 'rb') as f: | ||||
| h = hashlib.sha1(f.read()).hexdigest() | h = hashlib.sha1(f.read()).hexdigest() | ||||
| self.assertEqual(actual_sha1, h) | self.assertEqual(actual_sha1, h) | ||||
| self.assertNotEqual(actual_sha1, self.archive['sha1sum']) | self.assertNotEqual(actual_sha1, self.archive['sha1sum']) | ||||
| self.assertNotEqual(actual_sha1, self.archive2['sha1sum']) | self.assertNotEqual(actual_sha1, self.archive2['sha1sum']) | ||||
| @istest | def test_access_to_existing_deposit_with_multiple_archives(self): | ||||
| def access_to_existing_deposit_with_multiple_archives(self): | |||||
| """Access to deposit should stream a 200 response with its raw contents | """Access to deposit should stream a 200 response with its raw contents | ||||
| """ | """ | ||||
| deposit_id = self.create_complex_binary_deposit() | deposit_id = self.create_complex_binary_deposit() | ||||
| url = reverse(PRIVATE_GET_RAW_CONTENT, | url = reverse(PRIVATE_GET_RAW_CONTENT, | ||||
| args=[self.collection.name, deposit_id]) | args=[self.collection.name, deposit_id]) | ||||
| Show All 9 Lines | def test_access_to_existing_deposit_with_multiple_archives(self): | ||||
| # this touches the extraction directory but should clean up | # this touches the extraction directory but should clean up | ||||
| # after itself | # after itself | ||||
| self.assertEquals(os.listdir(TEST_CONFIG['extraction_dir']), []) | self.assertEquals(os.listdir(TEST_CONFIG['extraction_dir']), []) | ||||
| class DepositReadArchivesFailureTest(APITestCase, WithAuthTestCase, | class DepositReadArchivesFailureTest(APITestCase, WithAuthTestCase, | ||||
| BasicTestCase, CommonCreationRoutine): | BasicTestCase, CommonCreationRoutine): | ||||
| @istest | def test_access_to_nonexisting_deposit_returns_404_response(self): | ||||
| def access_to_nonexisting_deposit_returns_404_response(self): | |||||
| """Read unknown collection should return a 404 response | """Read unknown collection should return a 404 response | ||||
| """ | """ | ||||
| unknown_id = '999' | unknown_id = '999' | ||||
| url = reverse(PRIVATE_GET_RAW_CONTENT, | url = reverse(PRIVATE_GET_RAW_CONTENT, | ||||
| args=[self.collection.name, unknown_id]) | args=[self.collection.name, unknown_id]) | ||||
| response = self.client.get(url) | response = self.client.get(url) | ||||
| self.assertEqual(response.status_code, | self.assertEqual(response.status_code, | ||||
| status.HTTP_404_NOT_FOUND) | status.HTTP_404_NOT_FOUND) | ||||
| self.assertIn('Deposit with id %s does not exist' % unknown_id, | self.assertIn('Deposit with id %s does not exist' % unknown_id, | ||||
| response.content.decode('utf-8')) | response.content.decode('utf-8')) | ||||
| @istest | def test_access_to_nonexisting_collection_returns_404_response(self): | ||||
| def access_to_nonexisting_collection_returns_404_response(self): | |||||
| """Read unknown deposit should return a 404 response | """Read unknown deposit should return a 404 response | ||||
| """ | """ | ||||
| collection_name = 'non-existing' | collection_name = 'non-existing' | ||||
| deposit_id = self.create_deposit_partial() | deposit_id = self.create_deposit_partial() | ||||
| url = reverse(PRIVATE_GET_RAW_CONTENT, | url = reverse(PRIVATE_GET_RAW_CONTENT, | ||||
| args=[collection_name, deposit_id]) | args=[collection_name, deposit_id]) | ||||
| response = self.client.get(url) | response = self.client.get(url) | ||||
| self.assertEqual(response.status_code, | self.assertEqual(response.status_code, | ||||
| status.HTTP_404_NOT_FOUND) | status.HTTP_404_NOT_FOUND) | ||||
| self.assertIn('Unknown collection name %s' % collection_name, | self.assertIn('Unknown collection name %s' % collection_name, | ||||
| response.content.decode('utf-8')) | response.content.decode('utf-8')) | ||||