Differential D876 Diff 2795 swh/indexer/tests/test_utils.py

Changeset View

Standalone View

swh/indexer/tests/test_utils.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 abc
import datetime		import datetime
import hashlib		import hashlib
import random		import random

from swh.model import hashutil		from swh.model import hashutil
from swh.model.hashutil import hash_to_bytes, hash_to_hex		from swh.model.hashutil import hash_to_bytes, hash_to_hex

from swh.indexer.storage import INDEXER_CFG_KEY		from swh.indexer.storage import INDEXER_CFG_KEY
▲ Show 20 Lines • Show All 455 Lines • ▼ Show 20 Lines	class CommonIndexerWithErrorsTest:

def test_wrong_unknown_configuration_tool_range(self):		def test_wrong_unknown_configuration_tool_range(self):
"""Range Indexer with unknown configuration tool fails check"""		"""Range Indexer with unknown configuration tool fails check"""
if self.RangeIndexer is not None:		if self.RangeIndexer is not None:
with self.assertRaisesRegex(ValueError, 'Tools None is unknown'):		with self.assertRaisesRegex(ValueError, 'Tools None is unknown'):
self.RangeIndexer()		self.RangeIndexer()


class CommonContentIndexerTest:		class CommonContentIndexerTest(metaclass=abc.ABCMeta):
		legacy_get_format = False
		"""True iff the tested indexer uses the legacy format.
		see: https://forge.softwareheritage.org/T1433"""

def get_indexer_results(self, ids):		def get_indexer_results(self, ids):
		olasdUnsubmitted Not Done Inline Actions If you want to go that way, I'd rather make that attribute default to False, leaving legacy indexer test classes override it, than making it an abstract property with a hack. Did I ever tell you how I hate pretty much everything about abc? :-) olasd: If you want to go that way, I'd rather make that attribute default to False, leaving legacy…
"""Override this for indexers that don't have a mock storage."""		"""Override this for indexers that don't have a mock storage."""
return self.indexer.idx_storage.state		return self.indexer.idx_storage.state

def assert_results_ok(self, sha1s, expected_results=None):		def assert_legacy_results_ok(self, sha1s, expected_results=None):
		# XXX old format, remove this when all endpoints are
		# updated to the new one
		# see: https://forge.softwareheritage.org/T1433
sha1s = [sha1 if isinstance(sha1, bytes) else hash_to_bytes(sha1)		sha1s = [sha1 if isinstance(sha1, bytes) else hash_to_bytes(sha1)
for sha1 in sha1s]		for sha1 in sha1s]
actual_results = self.get_indexer_results(sha1s)		actual_results = list(self.get_indexer_results(sha1s))

if expected_results is None:		if expected_results is None:
expected_results = self.expected_results		expected_results = self.expected_results

		self.assertEqual(len(expected_results), len(actual_results),
		(expected_results, actual_results))
for indexed_data in actual_results:		for indexed_data in actual_results:
_id = indexed_data['id']		_id = indexed_data['id']
self.assertEqual(indexed_data, expected_results[_id])		expected_data = expected_results[hashutil.hash_to_hex(_id)].copy()
_tool_id = indexed_data['indexer_configuration_id']		expected_data['id'] = _id
		olasdUnsubmitted Done Inline Actions What's the task reference for the update to the new format? olasd: What's the task reference for the update to the new format?
		vlorentzAuthorUnsubmitted Done Inline Actions T1433 vlorentz: T1433
		olasdUnsubmitted Done Inline Actions I meant to reference it in the comment ;) olasd: I meant to reference it in the comment ;)
self.assertEqual(_tool_id, self.indexer.tool['id'])		self.assertEqual(indexed_data, expected_data)

		def assert_results_ok(self, sha1s, expected_results=None):
		if self.legacy_get_format:
		self.assert_legacy_results_ok(sha1s, expected_results)
		return

		sha1s = [sha1 if isinstance(sha1, bytes) else hash_to_bytes(sha1)
		for sha1 in sha1s]
		actual_results = list(self.get_indexer_results(sha1s))

		if expected_results is None:
		expected_results = self.expected_results

		self.assertEqual(len(expected_results), len(actual_results),
		(expected_results, actual_results))
		for indexed_data in actual_results:
		(_id, indexed_data) = list(indexed_data.items())[0]
		expected_data = expected_results[hashutil.hash_to_hex(_id)].copy()
		expected_data = [expected_data]
		self.assertEqual(indexed_data, expected_data)

def test_index(self):		def test_index(self):
"""Known sha1 have their data indexed		"""Known sha1 have their data indexed

"""		"""
sha1s = [self.id0, self.id1, self.id2]		sha1s = [self.id0, self.id1, self.id2]

# when		# when
▲ Show 20 Lines • Show All 135 Lines • Show Last 20 Lines