Differential D772 Diff 2424 swh/loader/core/tests/__init__.py

Changeset View

Standalone View

swh/loader/core/tests/init.py

	# Copyright (C) 2018 The Software Heritage developers			# Copyright (C) 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 functools
	import os			import os
	import pytest			import pytest
	import shutil			import shutil
	import subprocess			import subprocess
	import tempfile			import tempfile
				import warnings

	from unittest import TestCase			from unittest import TestCase

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


				def alias(new_method):
				@functools.wraps(new_method)
				def newf(self, args, *kwargs):
				# Find which method is myself, to infer my own name
				for attr_name in dir(self.__class__):
				if getattr(self.__class__, attr_name) is newf:
				old_name = attr_name
				break
				else:
				assert False, 'Result of alias() was not bound to a class.'

				message = '{old_name} is deprecated, please use {new_name} instead' \
				.format(old_name=old_name, new_name=new_method.__name__)
				warnings.warn(message, DeprecationWarning, stacklevel=2)

				return new_method(self, args, *kwargs)

				return newf


	class BaseLoaderStorageTest:			class BaseLoaderStorageTest:
	def _assertCountOk(self, type, expected_length, msg=None):			def _assertCountEqual(self, type, expected_length, msg=None):
	"""Check typed 'type' state to have the same expected length.			"""Check typed 'type' state to have the same expected length.

	"""			"""
	self.storage.refresh_stat_counters()			self.storage.refresh_stat_counters()
	self.assertEqual(self.storage.stat_counters()[type],			self.assertEqual(self.storage.stat_counters()[type],
	expected_length, msg=msg)			expected_length, msg=msg)

	def assertCountContents(self, len_expected_contents, msg=None):			def assertCountContents(self, len_expected_contents, msg=None):
	self._assertCountOk('content', len_expected_contents, msg=msg)			self._assertCountEqual('content', len_expected_contents, msg=msg)

	def assertCountDirectories(self, len_expected_directories, msg=None):			def assertCountDirectories(self, len_expected_directories, msg=None):
	self._assertCountOk('directory', len_expected_directories, msg=msg)			self._assertCountEqual('directory', len_expected_directories,
				msg=msg)

	def assertCountReleases(self, len_expected_releases, msg=None):			def assertCountReleases(self, len_expected_releases, msg=None):
	self._assertCountOk('release', len_expected_releases, msg=msg)			self._assertCountEqual('release', len_expected_releases, msg=msg)

	def assertCountRevisions(self, len_expected_revisions, msg=None):			def assertCountRevisions(self, len_expected_revisions, msg=None):
	self._assertCountOk('revision', len_expected_revisions, msg=msg)			self._assertCountEqual('revision', len_expected_revisions, msg=msg)

	def assertCountSnapshots(self, len_expected_snapshot, msg=None):			def assertCountSnapshots(self, len_expected_snapshot, msg=None):
	self._assertCountOk('snapshot', len_expected_snapshot, msg=msg)			self._assertCountEqual('snapshot', len_expected_snapshot, msg=msg)

	def assertContentsOk(self, expected_contents):			def assertContentsContain(self, expected_contents):
				anlambertUnsubmitted Not Done Inline Actions s/Checks/Check/ anlambert: s/Checks/Check/
	self._assertCountOk('content', len(expected_contents))			"""Checks the provided content are a subset of the stored ones.

				Args:
				expected_contents ([sha1]): List of content ids"""
				self._assertCountEqual('content', len(expected_contents))
	missing = list(self.storage.content_missing(			missing = list(self.storage.content_missing(
	{'sha1': hash_to_bytes(content_hash)}			{'sha1': hash_to_bytes(content_hash)}
	for content_hash in expected_contents))			for content_hash in expected_contents))
	self.assertEqual(missing, [])			self.assertEqual(missing, [])

	def assertDirectoriesOk(self, expected_directories):			assertContentsOk = alias(assertContentsContain)
	self._assertCountOk('directory', len(expected_directories))
				def assertDirectoriesContain(self, expected_directories):
				anlambertUnsubmitted Not Done Inline Actions s/Checks/Check/ anlambert: s/Checks/Check/
				"""Checks the provided directories are a subset of the stored ones.

				Args:
				expected_directories ([sha1]): List of directory ids."""
				self._assertCountEqual('directory', len(expected_directories))
	missing = list(self.storage.directory_missing(			missing = list(self.storage.directory_missing(
	dir_['id'] for dir_ in expected_directories))			dir_['id'] for dir_ in expected_directories))
	self.assertEqual(missing, [])			self.assertEqual(missing, [])

	def assertReleasesOk(self, expected_releases):			assertDirectoriesOk = alias(assertDirectoriesContain)
	"""Check the loader's releases match the expected releases.
				def assertReleasesContain(self, expected_releases):
				"""Checks the provided releases are a subset of the stored ones.

	Args:			Args:
	releases ([dict]): List of dictionaries representing swh releases.			releases ([dict]): List of dictionaries representing swh releases.

	"""			"""
	self._assertCountOk('release', len(expected_releases))			self._assertCountEqual('release', len(expected_releases))
	missing = list(self.storage.releases_missing(			missing = list(self.storage.releases_missing(
	rel['id'] for rel in expected_releases))			rel['id'] for rel in expected_releases))
	self.assertEqual(missing, [])			self.assertEqual(missing, [])

	def assertRevisionsOk(self, expected_revisions):			assertReleasesOk = alias(assertReleasesContain)
	"""Check the loader's revisions match the expected revisions.
				def assertRevisionsContain(self, expected_revisions):
				anlambertUnsubmitted Not Done Inline Actions s/Checks/Check/ anlambert: s/Checks/Check/
				"""Checks the provided revisions are a subset of the stored ones.

	Expects self.loader to be instantiated and ready to be			Expects self.loader to be instantiated and ready to be
	inspected (meaning the loading took place).			inspected (meaning the loading took place).

	Args:			Args:
	expected_revisions (dict): Dict with key revision id,			expected_revisions (dict): Dict with key revision id,
	value the targeted directory id.			value the targeted directory id.

	"""			"""
	self._assertCountOk('revision', len(expected_revisions))			self._assertCountEqual('revision', len(expected_revisions))

	revs = list(self.storage.revision_get(			revs = list(self.storage.revision_get(
	hashutil.hash_to_bytes(rev_id) for rev_id in expected_revisions))			hashutil.hash_to_bytes(rev_id) for rev_id in expected_revisions))
	self.assertNotIn(None, revs)			self.assertNotIn(None, revs)
	self.assertEqual(			self.assertEqual(
	{rev['id']: rev['directory'] for rev in revs},			{rev['id']: rev['directory'] for rev in revs},
	{hash_to_bytes(rev_id): hash_to_bytes(rev_dir)			{hash_to_bytes(rev_id): hash_to_bytes(rev_dir)
	for (rev_id, rev_dir) in expected_revisions.items()})			for (rev_id, rev_dir) in expected_revisions.items()})

	def assertSnapshotOk(self, expected_snapshot, expected_branches=[]):			assertRevisionsOk = alias(assertRevisionsContain)

				def assertSnapshotEqual(self, expected_snapshot, expected_branches=[]):
	"""Check for snapshot match.			"""Check for snapshot match.

	Provide the hashes as hexadecimal, the conversion is done			Provide the hashes as hexadecimal, the conversion is done
	within the method.			within the method.

	Args:			Args:

	expected_snapshot (str/dict): Either the snapshot			expected_snapshot (str/dict): Either the snapshot
	identifier or the full			identifier or the full
	snapshot			snapshot
	expected_branches (dict): expected branches or nothing is			expected_branches (dict): expected branches or nothing is
	the full snapshot is provided			the full snapshot is provided

	"""			"""
	if isinstance(expected_snapshot, dict) and not expected_branches:			if isinstance(expected_snapshot, dict) and not expected_branches:
	expected_snapshot_id = expected_snapshot['id']			expected_snapshot_id = expected_snapshot['id']
	expected_branches = expected_snapshot['branches']			expected_branches = expected_snapshot['branches']
	else:			else:
	expected_snapshot_id = expected_snapshot			expected_snapshot_id = expected_snapshot

	self._assertCountOk('snapshot', 1)			self._assertCountEqual('snapshot', 1)

	snap = self.storage.snapshot_get(hash_to_bytes(expected_snapshot_id))			snap = self.storage.snapshot_get(hash_to_bytes(expected_snapshot_id))
	self.assertIsNotNone(snap)			self.assertIsNotNone(snap)

	def decode_target(target):			def decode_target(target):
	if not target:			if not target:
	return target			return target
	target_type = target['target_type']			target_type = target['target_type']

	if target_type == 'alias':			if target_type == 'alias':
	decoded_target = target['target'].decode('utf-8')			decoded_target = target['target'].decode('utf-8')
	else:			else:
	decoded_target = hashutil.hash_to_hex(target['target'])			decoded_target = hashutil.hash_to_hex(target['target'])

	return {			return {
	'target': decoded_target,			'target': decoded_target,
	'target_type': target_type			'target_type': target_type
	}			}

	branches = {			branches = {
	branch.decode('utf-8'): decode_target(target)			branch.decode('utf-8'): decode_target(target)
	for branch, target in snap['branches'].items()			for branch, target in snap['branches'].items()
	}			}
	self.assertEqual(expected_branches, branches)			self.assertEqual(expected_branches, branches)

				assertSnapshotOk = alias(assertSnapshotEqual)


				anlambertUnsubmitted Not Done Inline Actions s/Checks/Check/ anlambert: s/Checks/Check/
	@pytest.mark.fs			@pytest.mark.fs
	class BaseLoaderTest(TestCase, BaseLoaderStorageTest):			class BaseLoaderTest(TestCase, BaseLoaderStorageTest):
	"""Mixin base loader test class.			"""Mixin base loader test class.

	This allows to uncompress archives (mercurial, svn, git,			This allows to uncompress archives (mercurial, svn, git,
	... repositories) into a temporary folder so that the loader under			... repositories) into a temporary folder so that the loader under
	test can work with this.			test can work with this.

	▲ Show 20 Lines • Show All 56 Lines • Show Last 20 Lines