Differential D796 Diff 2537 swh/loader/tar/tests/test_loader.py

Changeset View

Standalone View

swh/loader/tar/tests/test_loader.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 os		import os

import pytest		import pytest
		import requests_mock

from swh.model import hashutil		from swh.model import hashutil

from swh.loader.core.tests import BaseLoaderTest		from swh.loader.core.tests import BaseLoaderTest
from swh.loader.tar.build import SWH_PERSON		from swh.loader.tar.build import SWH_PERSON
from swh.loader.tar.loader import RemoteTarLoader, TarLoader		from swh.loader.tar.loader import RemoteTarLoader, TarLoader


Show All 20 Lines
}		}


class RemoteTarLoaderForTest(RemoteTarLoader):		class RemoteTarLoaderForTest(RemoteTarLoader):
def parse_config_file(self, args, *kwargs):		def parse_config_file(self, args, *kwargs):
return TEST_CONFIG		return TEST_CONFIG


class PrepareDataForTestLoader(BaseLoaderTest):		class PrepareDataForTestLoader(BaseLoaderTest):
		vlorentzUnsubmitted Not Done Inline Actions Should be decorated with `@pytest.mark.fs` vlorentz: Should be decorated with `@pytest.mark.fs`
"""Prepare the archive to load (test fixture).		"""Prepare the archive to load (test fixture).

"""		"""
def setUp(self):		def setUp(self):
super().setUp('sample-folder.tgz',		super().setUp('sample-folder.tgz',
start_path=os.path.dirname(__file__),		start_path=os.path.dirname(__file__),
uncompress_archive=False)		uncompress_archive=False)
self.tarpath = self.destination_path		self.tarpath = self.destination_path

		def assert_data_ok(self):
class TestTarLoader1(PrepareDataForTestLoader):
"""Test the remote loader

"""
def setUp(self):
super().setUp()
self.loader = RemoteTarLoaderForTest()
self.storage = self.loader.storage

@pytest.mark.fs
def test_load(self):
"""Process a new tarball should be ok

"""
# given
origin = {
'url': self.repo_url,
'type': 'tar'
}

visit_date = 'Tue, 3 May 2016 17:16:32 +0200'

last_modified = '2018-12-05T12:35:23+00:00'

# when
self.loader.load(
origin=origin, visit_date=visit_date, last_modified=last_modified)

# then		# then
self.assertCountContents(8, "3 files + 5 links")		self.assertCountContents(8, "3 files + 5 links")
self.assertCountDirectories(6, "4 subdirs + 1 empty + 1 main dir")		self.assertCountDirectories(6, "4 subdirs + 1 empty + 1 main dir")
self.assertCountRevisions(1, "synthetic revision")		self.assertCountRevisions(1, "synthetic revision")

rev_id = hashutil.hash_to_bytes(		rev_id = hashutil.hash_to_bytes(
'67a7d7dda748f9a86b56a13d9218d16f5cc9ab3d')		'67a7d7dda748f9a86b56a13d9218d16f5cc9ab3d')
actual_revision = next(self.storage.revision_get([rev_id]))		actual_revision = next(self.storage.revision_get([rev_id]))
Show All 16 Lines	def assert_data_ok(self):
'length': 555,		'length': 555,
'sha256': '307ebda0071ca5975f618e192c8417161e19b6c8bf581a26061b76dc8e85321d' # noqa		'sha256': '307ebda0071ca5975f618e192c8417161e19b6c8bf581a26061b76dc8e85321d' # noqa
})		})

self.assertCountReleases(0)		self.assertCountReleases(0)
self.assertCountSnapshots(1)		self.assertCountSnapshots(1)


		class TestRemoteTarLoader(PrepareDataForTestLoader):
		"""Test the remote loader scenario (local/remote)

		"""
		def setUp(self):
		super().setUp()
		self.loader = RemoteTarLoaderForTest()
		self.storage = self.loader.storage

		vlorentzUnsubmitted Not Done Inline Actions Remove that decorator from the test and move it to the class (all tests of the class use the FS) vlorentz: Remove that decorator from the test and move it to the class (all tests of the class use the FS)
		@pytest.mark.fs
		def test_load_local(self):
		"""Load a local tarball should result in persisted swh data

		"""
		# given
		origin = {
		'url': self.repo_url,
		'type': 'tar'
		}

		visit_date = 'Tue, 3 May 2016 17:16:32 +0200'

		last_modified = '2018-12-05T12:35:23+00:00'

		# when
		self.loader.load(
		origin=origin, visit_date=visit_date, last_modified=last_modified)

		# then
		self.assert_data_ok()

		@requests_mock.Mocker()
		def test_load_remote(self, mock_requests):
		"""Load a remote tarball should result in persisted swh data

		"""
		# setup the mock to stream the content of the tarball
		local_url = self.repo_url.replace('file:///', '/')
		url = 'https://nowhere.org/%s' % local_url
		with open(local_url, 'rb') as f:
		data = f.read()
		mock_requests.get(url, content=data, headers={
		'content-length': str(len(data))
		})

		# given
		origin = {
		'url': url,
		'type': 'tar'
		}

		visit_date = 'Tue, 3 May 2016 17:16:32 +0200'

		last_modified = '2018-12-05T12:35:23+00:00'

		# when
		self.loader.load(
		origin=origin, visit_date=visit_date, last_modified=last_modified)

		self.assert_data_ok()

		@requests_mock.Mocker()
		def test_load_remote_download_failure(self, mock_requests):
		"""Load a remote tarball with download failure should result in no data

		"""
		# setup the mock to stream the content of the tarball
		local_url = self.repo_url.replace('file:///', '/')
		url = 'https://nowhere.org/%s' % local_url
		with open(local_url, 'rb') as f:
		data = f.read()
		wrong_length = len(data) - 10
		mock_requests.get(url, content=data, headers={
		'content-length': str(wrong_length)
		})

		# given
		origin = {
		'url': url,
		'type': 'tar'
		}

		visit_date = 'Tue, 3 May 2016 17:16:32 +0200'

		last_modified = '2018-12-05T12:35:23+00:00'

		# when
		r = self.loader.load(
		origin=origin, visit_date=visit_date,
		last_modified=last_modified)

		self.assertEqual(r, {'status': 'failed'})
		self.assertCountContents(0)
		self.assertCountDirectories(0)
		self.assertCountRevisions(0)
		self.assertCountSnapshots(0)


class TarLoaderForTest(TarLoader):		class TarLoaderForTest(TarLoader):
def parse_config_file(self, args, *kwargs):		def parse_config_file(self, args, *kwargs):
return TEST_CONFIG		return TEST_CONFIG


class TestTarLoader2(PrepareDataForTestLoader):		class TestTarLoader2(PrepareDataForTestLoader):
"""Test the legacy tar loader		"""Test the legacy tar loader

▲ Show 20 Lines • Show All 44 Lines • ▼ Show 20 Lines	def test_load(self):
branch_name = os.path.basename(self.tarpath)		branch_name = os.path.basename(self.tarpath)

# when		# when
self.loader.load(tar_path=self.tarpath, origin=origin,		self.loader.load(tar_path=self.tarpath, origin=origin,
visit_date=visit_date, revision=revision,		visit_date=visit_date, revision=revision,
branch_name=branch_name)		branch_name=branch_name)

# then		# then
self.assertCountContents(8, "3 files + 5 links")		self.assert_data_ok()
self.assertCountDirectories(6, "4 subdirs + 1 empty + 1 main dir")
self.assertCountRevisions(1, "synthetic revision")

rev_id = hashutil.hash_to_bytes(
'67a7d7dda748f9a86b56a13d9218d16f5cc9ab3d')
actual_revision = next(self.storage.revision_get([rev_id]))
self.assertTrue(actual_revision['synthetic'])
self.assertEqual(actual_revision['parents'], [])
self.assertEqual(actual_revision['type'], 'tar')
self.assertEqual(actual_revision['message'],
b'swh-loader-tar: synthetic revision message')
self.assertEqual(actual_revision['directory'],
b'\xa7A\xfcM\x96\x8c{\x8e<\x94\xff\x86\xe7\x04\x80\xc5\xc7\xe5r\xa9') # noqa

self.assertEqual(
actual_revision['metadata']['original_artifact'][0],
{
'sha1_git': 'cc848944a0d3e71d287027347e25467e61b07428',
'archive_type': 'tar',
'blake2s256': '5d70923443ad36377cd58e993aff0e3c1b9ef14f796c69569105d3a99c64f075', # noqa
'name': 'sample-folder.tgz',
'sha1': '3ca0d0a5c6833113bd532dc5c99d9648d618f65a',
'length': 555,
'sha256': '307ebda0071ca5975f618e192c8417161e19b6c8bf581a26061b76dc8e85321d' # noqa
})

self.assertCountReleases(0)
self.assertCountSnapshots(1)