Differential D1095 Diff 3438 swh/indexer/tests/storage/test_storage.py

Changeset View

Standalone View

swh/indexer/tests/storage/test_storage.py

Show First 20 Lines • Show All 92 Lines • ▼ Show 20 Lines	class BasePgTestStorage(SingleDbTestFixture):
def reset_storage_tables(self):		def reset_storage_tables(self):
excluded = {'indexer_configuration'}		excluded = {'indexer_configuration'}
self.reset_db_tables(self.TEST_DB_NAME, excluded=excluded)		self.reset_db_tables(self.TEST_DB_NAME, excluded=excluded)

db = self.test_db[self.TEST_DB_NAME]		db = self.test_db[self.TEST_DB_NAME]
db.conn.commit()		db.conn.commit()


class CommonTestStorage:		def gen_generic_endpoint_tests(endpoint_type, tool_name,
"""Base class for Indexer Storage testing.		example_data1, example_data2):
		def rename(f):
		f.__name__ = 'test_' + endpoint_type + f.__name__
		return f

"""		def endpoint(self, endpoint_name):
def setUp(self):		return getattr(self.storage, endpoint_type + '_' + endpoint_name)
super().setUp()
self.storage = get_indexer_storage(**self.storage_config)
tools = self.storage.indexer_configuration_add(TOOLS)
self.tools = {}
for tool in tools:
tool_name = tool['tool_name']
while tool_name in self.tools:
tool_name += '_'
self.tools[tool_name] = {
'id': tool['id'],
'name': tool['tool_name'],
'version': tool['tool_version'],
'configuration': tool['tool_configuration'],
}

self.sha1_1 = hash_to_bytes('34973274ccef6ab4dfaaf86599792fa9c3fe4689')		@rename
self.sha1_2 = hash_to_bytes('61c2b3a30496d329e21af70dd2d7e097046d07b7')		def missing(self):
self.revision_id_1 = hash_to_bytes(
'7026b7c1a2af56521e951c01ed20f255fa054238')
self.revision_id_2 = hash_to_bytes(
'7026b7c1a2af56521e9587659012345678904321')
self.origin_id_1 = 54974445
self.origin_id_2 = 44434342

def test_check_config(self):
self.assertTrue(self.storage.check_config(check_write=True))
self.assertTrue(self.storage.check_config(check_write=False))

def test_content_mimetype_missing(self):
# given		# given
tool_id = self.tools['file']['id']		tool_id = self.tools[tool_name]['id']

mimetypes = [		query = [
{		{
'id': self.sha1_1,		'id': self.sha1_1,
'indexer_configuration_id': tool_id,		'indexer_configuration_id': tool_id,
},		},
{		{
'id': self.sha1_2,		'id': self.sha1_2,
'indexer_configuration_id': tool_id,		'indexer_configuration_id': tool_id,
}]		}]

# when		# when
actual_missing = self.storage.content_mimetype_missing(mimetypes)		actual_missing = endpoint(self, 'missing')(query)

# then		# then
self.assertEqual(list(actual_missing), [		self.assertEqual(list(actual_missing), [
self.sha1_1,		self.sha1_1,
self.sha1_2,		self.sha1_2,
])		])

# given		# given
self.storage.content_mimetype_add([{		endpoint(self, 'add')([{
'id': self.sha1_2,		'id': self.sha1_2,
'mimetype': 'text/plain',		**example_data1,
'encoding': 'utf-8',
'indexer_configuration_id': tool_id,		'indexer_configuration_id': tool_id,
}])		}])

# when		# when
actual_missing = self.storage.content_mimetype_missing(mimetypes)		actual_missing = endpoint(self, 'missing')(query)

# then		# then
self.assertEqual(list(actual_missing), [self.sha1_1])		self.assertEqual(list(actual_missing), [self.sha1_1])

def test_content_mimetype_add__drop_duplicate(self):		@rename
		def add__drop_duplicate(self):
# given		# given
tool_id = self.tools['file']['id']		tool_id = self.tools[tool_name]['id']

mimetype_v1 = {		data_v1 = {
'id': self.sha1_2,		'id': self.sha1_2,
'mimetype': 'text/plain',		**example_data1,
'encoding': 'utf-8',
'indexer_configuration_id': tool_id,		'indexer_configuration_id': tool_id,
}		}

# given		# given
self.storage.content_mimetype_add([mimetype_v1])		endpoint(self, 'add')([data_v1])

# when		# when
actual_mimetypes = list(self.storage.content_mimetype_get(		actual_data = list(endpoint(self, 'get')([self.sha1_2]))
[self.sha1_2]))

# then		# then
expected_mimetypes_v1 = [{		expected_data_v1 = [{
'id': self.sha1_2,		'id': self.sha1_2,
'mimetype': 'text/plain',		**example_data1,
'encoding': 'utf-8',		'tool': self.tools[tool_name],
'tool': self.tools['file'],
}]		}]
self.assertEqual(actual_mimetypes, expected_mimetypes_v1)		self.assertEqual(actual_data, expected_data_v1)

# given		# given
mimetype_v2 = mimetype_v1.copy()		data_v2 = data_v1.copy()
mimetype_v2.update({		data_v2.update(example_data2)
'mimetype': 'text/html',
'encoding': 'us-ascii',
})

self.storage.content_mimetype_add([mimetype_v2])		endpoint(self, 'add')([data_v2])

actual_mimetypes = list(self.storage.content_mimetype_get(		actual_data = list(endpoint(self, 'get')([self.sha1_2]))
[self.sha1_2]))

# mimetype did not change as the v2 was dropped.		# data did not change as the v2 was dropped.
self.assertEqual(actual_mimetypes, expected_mimetypes_v1)		self.assertEqual(actual_data, expected_data_v1)

def test_content_mimetype_add__update_in_place_duplicate(self):		@rename
		def add__update_in_place_duplicate(self):
# given		# given
tool_id = self.tools['file']['id']		tool_id = self.tools[tool_name]['id']

mimetype_v1 = {		data_v1 = {
'id': self.sha1_2,		'id': self.sha1_2,
'mimetype': 'text/plain',		**example_data1,
'encoding': 'utf-8',
'indexer_configuration_id': tool_id,		'indexer_configuration_id': tool_id,
}		}

# given		# given
self.storage.content_mimetype_add([mimetype_v1])		endpoint(self, 'add')([data_v1])

# when		# when
actual_mimetypes = list(self.storage.content_mimetype_get(		actual_data = list(endpoint(self, 'get')([self.sha1_2]))
[self.sha1_2]))

expected_mimetypes_v1 = [{		expected_data_v1 = [{
'id': self.sha1_2,		'id': self.sha1_2,
'mimetype': 'text/plain',		**example_data1,
'encoding': 'utf-8',		'tool': self.tools[tool_name],
'tool': self.tools['file'],
}]		}]

# then		# then
self.assertEqual(actual_mimetypes, expected_mimetypes_v1)		self.assertEqual(actual_data, expected_data_v1)

# given		# given
mimetype_v2 = mimetype_v1.copy()		data_v2 = data_v1.copy()
mimetype_v2.update({		data_v2.update(example_data2)
'mimetype': 'text/html',
'encoding': 'us-ascii',
})

self.storage.content_mimetype_add([mimetype_v2], conflict_update=True)		endpoint(self, 'add')([data_v2], conflict_update=True)

actual_mimetypes = list(self.storage.content_mimetype_get(		actual_data = list(endpoint(self, 'get')([self.sha1_2]))
[self.sha1_2]))

expected_mimetypes_v2 = [{		expected_data_v2 = [{
'id': self.sha1_2,		'id': self.sha1_2,
'mimetype': 'text/html',		**example_data2,
'encoding': 'us-ascii',		'tool': self.tools[tool_name],
'tool': {
'id': self.tools['file']['id'],
'name': 'file',
'version': '5.22',
'configuration': {'command_line': 'file --mime <filepath>'}
}
}]		}]

# mimetype did change as the v2 was used to overwrite v1		# data did change as the v2 was used to overwrite v1
self.assertEqual(actual_mimetypes, expected_mimetypes_v2)		self.assertEqual(actual_data, expected_data_v2)

def test_content_mimetype_get(self):		@rename
		def get(self):
# given		# given
tool_id = self.tools['file']['id']		tool_id = self.tools[tool_name]['id']

mimetypes = [self.sha1_2, self.sha1_1]		query = [self.sha1_2, self.sha1_1]

mimetype1 = {		data1 = {
'id': self.sha1_2,		'id': self.sha1_2,
'mimetype': 'text/plain',		**example_data1,
'encoding': 'utf-8',
'indexer_configuration_id': tool_id,		'indexer_configuration_id': tool_id,
}		}

# when		# when
self.storage.content_mimetype_add([mimetype1])		endpoint(self, 'add')([data1])

# then		# then
actual_mimetypes = list(self.storage.content_mimetype_get(mimetypes))		actual_data = list(endpoint(self, 'get')(query))

# then		# then
expected_mimetypes = [{		expected_data = [{
'id': self.sha1_2,		'id': self.sha1_2,
'mimetype': 'text/plain',		**example_data1,
'encoding': 'utf-8',		'tool': self.tools[tool_name]
'tool': self.tools['file']
}]		}]

self.assertEqual(actual_mimetypes, expected_mimetypes)		self.assertEqual(actual_data, expected_data)

def test_content_language_missing(self):
# given
tool_id = self.tools['pygments']['id']

languages = [
{
'id': self.sha1_2,
'indexer_configuration_id': tool_id,
},
{
'id': self.sha1_1,
'indexer_configuration_id': tool_id,
}
]

# when
actual_missing = list(self.storage.content_language_missing(languages))

# then
self.assertEqual(list(actual_missing), [
self.sha1_2,
self.sha1_1,
])

# given		return (
self.storage.content_language_add([{		missing,
'id': self.sha1_2,		add__drop_duplicate,
'lang': 'haskell',		add__update_in_place_duplicate,
'indexer_configuration_id': tool_id,		get,
}])		)

# when
actual_missing = list(self.storage.content_language_missing(languages))

# then
self.assertEqual(actual_missing, [self.sha1_1])

def test_content_language_get(self):
# given
tool_id = self.tools['pygments']['id']

language1 = {
'id': self.sha1_2,
'lang': 'common-lisp',
'indexer_configuration_id': tool_id,
}

# when
self.storage.content_language_add([language1])

# then
actual_languages = list(self.storage.content_language_get(
[self.sha1_2, self.sha1_1]))

# then
expected_languages = [{
'id': self.sha1_2,
'lang': 'common-lisp',
'tool': self.tools['pygments']
}]

self.assertEqual(actual_languages, expected_languages)

def test_content_language_add__drop_duplicate(self):		class CommonTestStorage:
# given		"""Base class for Indexer Storage testing.
tool_id = self.tools['pygments']['id']

language_v1 = {		"""
'id': self.sha1_2,		def setUp(self):
'lang': 'emacslisp',		super().setUp()
'indexer_configuration_id': tool_id,		self.storage = get_indexer_storage(**self.storage_config)
		tools = self.storage.indexer_configuration_add(TOOLS)
		self.tools = {}
		for tool in tools:
		tool_name = tool['tool_name']
		while tool_name in self.tools:
		tool_name += '_'
		self.tools[tool_name] = {
		'id': tool['id'],
		'name': tool['tool_name'],
		'version': tool['tool_version'],
		'configuration': tool['tool_configuration'],
}		}

# given		self.sha1_1 = hash_to_bytes('34973274ccef6ab4dfaaf86599792fa9c3fe4689')
self.storage.content_language_add([language_v1])		self.sha1_2 = hash_to_bytes('61c2b3a30496d329e21af70dd2d7e097046d07b7')
		self.revision_id_1 = hash_to_bytes(
# when		'7026b7c1a2af56521e951c01ed20f255fa054238')
actual_languages = list(self.storage.content_language_get(		self.revision_id_2 = hash_to_bytes(
[self.sha1_2]))		'7026b7c1a2af56521e9587659012345678904321')
		self.origin_id_1 = 54974445
# then		self.origin_id_2 = 44434342
expected_languages_v1 = [{
'id': self.sha1_2,
'lang': 'emacslisp',
'tool': self.tools['pygments']
}]
self.assertEqual(actual_languages, expected_languages_v1)

# given
language_v2 = language_v1.copy()
language_v2.update({
'lang': 'common-lisp',
})

self.storage.content_language_add([language_v2])

actual_languages = list(self.storage.content_language_get(
[self.sha1_2]))

# language did not change as the v2 was dropped.
self.assertEqual(actual_languages, expected_languages_v1)

def test_content_language_add__update_in_place_duplicate(self):
# given
tool_id = self.tools['pygments']['id']

language_v1 = {
'id': self.sha1_2,
'lang': 'common-lisp',
'indexer_configuration_id': tool_id,
}

# given		def test_check_config(self):
self.storage.content_language_add([language_v1])		self.assertTrue(self.storage.check_config(check_write=True))
		self.assertTrue(self.storage.check_config(check_write=False))

# when		# generate content_mimetype tests
actual_languages = list(self.storage.content_language_get(		(
[self.sha1_2]))		test_content_mimetype_missing,
		test_content_mimetype_add__drop_duplicate,
		test_content_mimetype_add__update_in_place_duplicate,
		test_content_mimetype_get,
		) = gen_generic_endpoint_tests(
		endpoint_type='content_mimetype',
		tool_name='file',
		example_data1={
		'mimetype': 'text/plain',
		'encoding': 'utf-8',
		},
		example_data2={
		'mimetype': 'text/html',
		'encoding': 'us-ascii',
		},
		)

# then		# content_language tests
expected_languages_v1 = [{		(
'id': self.sha1_2,		test_content_language_missing,
		test_content_language_add__drop_duplicate,
		test_content_language_add__update_in_place_duplicate,
		test_content_language_get,
		) = gen_generic_endpoint_tests(
		endpoint_type='content_language',
		tool_name='pygments',
		example_data1={
		'lang': 'haskell',
		},
		example_data2={
'lang': 'common-lisp',		'lang': 'common-lisp',
'tool': self.tools['pygments']
}]
self.assertEqual(actual_languages, expected_languages_v1)

# given
language_v2 = language_v1.copy()
language_v2.update({
'lang': 'emacslisp',
})

self.storage.content_language_add([language_v2], conflict_update=True)

actual_languages = list(self.storage.content_language_get(
[self.sha1_2]))

# language did not change as the v2 was dropped.
expected_languages_v2 = [{
'id': self.sha1_2,
'lang': 'emacslisp',
'tool': self.tools['pygments']
}]

# language did change as the v2 was used to overwrite v1
self.assertEqual(actual_languages, expected_languages_v2)

def test_content_ctags_missing(self):
# given
tool_id = self.tools['universal-ctags']['id']

ctags = [
{
'id': self.sha1_2,
'indexer_configuration_id': tool_id,
},		},
{		)
'id': self.sha1_1,
'indexer_configuration_id': tool_id,
}
]

# when
actual_missing = self.storage.content_ctags_missing(ctags)

# then
self.assertEqual(list(actual_missing), [
self.sha1_2,
self.sha1_1
])

# given		# content_ctags tests
self.storage.content_ctags_add([		(
{		test_content_ctags_missing,
'id': self.sha1_2,		# the following tests are disabled because CTAGS behave differently
'indexer_configuration_id': tool_id,		_, # test_content_ctags_add__drop_duplicate,
		_, # test_content_ctags_add__update_in_place_duplicate,
		_, # test_content_ctags_get,
		) = gen_generic_endpoint_tests(
		endpoint_type='content_ctags',
		ardumontUnsubmitted Done Inline Actions It seems we lose coverage here. Do not migrate those tests to the new way and avoid losing coverage. ardumont: It seems we lose coverage here. Do not migrate those tests to the new way and avoid losing…
		ardumontUnsubmitted Done Inline Actions vlorentz: No we don't. I preserved tests that I could not factorize easily. cool ardumont: > vlorentz: No we don't. I preserved tests that I could not factorize easily. cool
		tool_name='universal-ctags',
		example_data1={
'ctags': [{		'ctags': [{
'name': 'done',		'name': 'done',
'kind': 'variable',		'kind': 'variable',
'line': 119,		'line': 119,
'lang': 'OCaml',		'lang': 'OCaml',
}]		}]
},		},
])		example_data2={

# when
actual_missing = self.storage.content_ctags_missing(ctags)

# then
self.assertEqual(list(actual_missing), [self.sha1_1])

def test_content_ctags_get(self):
# given
tool_id = self.tools['universal-ctags']['id']

ctags = [self.sha1_2, self.sha1_1]

ctag1 = {
'id': self.sha1_2,
'indexer_configuration_id': tool_id,
'ctags': [		'ctags': [
{		{
'name': 'done',		'name': 'done',
'kind': 'variable',		'kind': 'variable',
'line': 100,		'line': 100,
'lang': 'Python',		'lang': 'Python',
},		},
{		{
'name': 'main',		'name': 'main',
'kind': 'function',		'kind': 'function',
'line': 119,		'line': 119,
'lang': 'Python',		'lang': 'Python',
}]		}]
}

# when
self.storage.content_ctags_add([ctag1])

# then
actual_ctags = list(self.storage.content_ctags_get(ctags))

# then

expected_ctags = [
{
'id': self.sha1_2,
'tool': self.tools['universal-ctags'],
'name': 'done',
'kind': 'variable',
'line': 100,
'lang': 'Python',
},		},
{		)
'id': self.sha1_2,
'tool': self.tools['universal-ctags'],
'name': 'main',
'kind': 'function',
'line': 119,
'lang': 'Python',
}
]

self.assertEqual(actual_ctags, expected_ctags)

def test_content_ctags_search(self):		def test_content_ctags_search(self):
# 1. given		# 1. given
tool = self.tools['universal-ctags']		tool = self.tools['universal-ctags']
tool_id = tool['id']		tool_id = tool['id']

ctag1 = {		ctag1 = {
'id': self.sha1_1,		'id': self.sha1_1,
▲ Show 20 Lines • Show All 288 Lines • ▼ Show 20 Lines	def test_content_ctags_add__update_in_place(self):
'kind': 'function',		'kind': 'function',
'line': 120,		'line': 120,
'lang': 'Scheme',		'lang': 'Scheme',
'tool': tool,		'tool': tool,
}		}
]		]
self.assertEqual(actual_ctags, expected_ctags)		self.assertEqual(actual_ctags, expected_ctags)

def test_content_fossology_license_get(self):		# content_fossology_license tests
# given		(
tool = self.tools['nomos']		_, # The endpoint content_fossology_license_missing does not exist
tool_id = tool['id']		# the following tests are disabled because fossology_license tests
		# behave differently
license1 = {		_, # test_content_fossology_license_add__drop_duplicate,
'id': self.sha1_1,		_, # test_content_fossology_license_add__update_in_place_duplicate,
'licenses': ['GPL-2.0+'],		_, # test_content_fossology_license_get,
'indexer_configuration_id': tool_id,		) = gen_generic_endpoint_tests(
}		endpoint_type='content_fossology_license',
		tool_name='nomos',
# when		example_data1={
self.storage.content_fossology_license_add([license1])		'licenses': ['Apache-2.0'],
		},
# then		example_data2={
actual_licenses = list(self.storage.content_fossology_license_get(		'licenses': ['BSD-2-Clause'],
[self.sha1_2, self.sha1_1]))		},
		)
expected_license = {
self.sha1_1: [{
'licenses': ['GPL-2.0+'],
'tool': tool,
}]
}

# then
self.assertEqual(actual_licenses, [expected_license])

def test_content_fossology_license_add__new_license_added(self):		def test_content_fossology_license_add__new_license_added(self):
# given		# given
tool = self.tools['nomos']		tool = self.tools['nomos']
tool_id = tool['id']		tool_id = tool['id']

license_v1 = {		license_v1 = {
'id': self.sha1_1,		'id': self.sha1_1,
Show All 35 Lines	def test_content_fossology_license_add__new_license_added(self):
'licenses': ['Apache-2.0', 'BSD-2-Clause'],		'licenses': ['Apache-2.0', 'BSD-2-Clause'],
'tool': tool		'tool': tool
}]		}]
}		}

# license did not change as the v2 was dropped.		# license did not change as the v2 was dropped.
self.assertEqual(actual_licenses, [expected_license])		self.assertEqual(actual_licenses, [expected_license])

def test_content_fossology_license_add__update_in_place_duplicate(self):		# content_metadata tests
# given		(
tool = self.tools['nomos']		test_content_metadata_missing,
tool_id = tool['id']		test_content_metadata_add__drop_duplicate,
		test_content_metadata_add__update_in_place_duplicate,
license_v1 = {		test_content_metadata_get,
'id': self.sha1_1,		) = gen_generic_endpoint_tests(
'licenses': ['CECILL'],		endpoint_type='content_metadata',
'indexer_configuration_id': tool_id,		tool_name='swh-metadata-detector',
}		example_data1={

# given
self.storage.content_fossology_license_add([license_v1])
# conflict does nothing
self.storage.content_fossology_license_add([license_v1])

# when
actual_licenses = list(self.storage.content_fossology_license_get(
[self.sha1_1]))

# then
expected_license = {
self.sha1_1: [{
'licenses': ['CECILL'],
'tool': tool,
}]
}
self.assertEqual(actual_licenses, [expected_license])

# given
license_v2 = license_v1.copy()
license_v2.update({
'licenses': ['CECILL-2.0']
})

self.storage.content_fossology_license_add([license_v2],
conflict_update=True)

actual_licenses = list(self.storage.content_fossology_license_get(
[self.sha1_1]))

# license did change as the v2 was used to overwrite v1
expected_license = {
self.sha1_1: [{
'licenses': ['CECILL-2.0'],
'tool': tool,
}]
}
self.assertEqual(actual_licenses, [expected_license])

def test_content_metadata_missing(self):
# given
tool_id = self.tools['swh-metadata-translator']['id']

metadata = [
{
'id': self.sha1_2,
'indexer_configuration_id': tool_id,
},
{
'id': self.sha1_1,
'indexer_configuration_id': tool_id,
}
]

# when
actual_missing = list(self.storage.content_metadata_missing(metadata))

# then
self.assertEqual(list(actual_missing), [
self.sha1_2,
self.sha1_1,
])

# given
self.storage.content_metadata_add([{
'id': self.sha1_2,
'translated_metadata': {		'translated_metadata': {
'other': {},		'other': {},
'codeRepository': {		'codeRepository': {
'type': 'git',		'type': 'git',
'url': 'https://github.com/moranegg/metadata_test'		'url': 'https://github.com/moranegg/metadata_test'
},		},
'description': 'Simple package.json test for indexer',		'description': 'Simple package.json test for indexer',
'name': 'test_metadata',		'name': 'test_metadata',
'version': '0.0.1'		'version': '0.0.1'
},		},
'indexer_configuration_id': tool_id		},
}])		example_data2={

# when
actual_missing = list(self.storage.content_metadata_missing(metadata))

# then
self.assertEqual(actual_missing, [self.sha1_1])

def test_content_metadata_get(self):
# given
tool_id = self.tools['swh-metadata-translator']['id']

metadata1 = {
'id': self.sha1_2,
'translated_metadata': {		'translated_metadata': {
'other': {},		'other': {},
'codeRepository': {
'type': 'git',
'url': 'https://github.com/moranegg/metadata_test'
},
'description': 'Simple package.json test for indexer',
'name': 'test_metadata',		'name': 'test_metadata',
'version': '0.0.1'		'version': '0.0.1'
},		},
'indexer_configuration_id': tool_id,		},
}		)

# when
self.storage.content_metadata_add([metadata1])

# then
actual_metadata = list(self.storage.content_metadata_get(
[self.sha1_2, self.sha1_1]))

expected_metadata = [{		# revision_metadata tests
'id': self.sha1_2,		(
		test_revision_metadata_missing,
		test_revision_metadata_add__drop_duplicate,
		test_revision_metadata_add__update_in_place_duplicate,
		test_revision_metadata_get,
		) = gen_generic_endpoint_tests(
		endpoint_type='revision_metadata',
		tool_name='swh-metadata-detector',
		example_data1={
'translated_metadata': {		'translated_metadata': {
'other': {},		'other': {},
'codeRepository': {		'codeRepository': {
'type': 'git',		'type': 'git',
'url': 'https://github.com/moranegg/metadata_test'		'url': 'https://github.com/moranegg/metadata_test'
},		},
'description': 'Simple package.json test for indexer',		'description': 'Simple package.json test for indexer',
'name': 'test_metadata',		'name': 'test_metadata',
'version': '0.0.1'		'version': '0.0.1'
},		},
'tool': self.tools['swh-metadata-translator']		'mappings': ['mapping1'],
}]

self.assertEqual(actual_metadata, expected_metadata)

def test_content_metadata_add_drop_duplicate(self):
# given
tool_id = self.tools['swh-metadata-translator']['id']

metadata_v1 = {
'id': self.sha1_2,
'translated_metadata': {
'other': {},
'name': 'test_metadata',
'version': '0.0.1'
},
'indexer_configuration_id': tool_id,
}

# given
self.storage.content_metadata_add([metadata_v1])

# when
actual_metadata = list(self.storage.content_metadata_get(
[self.sha1_2]))

expected_metadata_v1 = [{
'id': self.sha1_2,
'translated_metadata': {
'other': {},
'name': 'test_metadata',
'version': '0.0.1'
},
'tool': self.tools['swh-metadata-translator']
}]

self.assertEqual(actual_metadata, expected_metadata_v1)

# given
metadata_v2 = metadata_v1.copy()
metadata_v2.update({
'translated_metadata': {
'other': {},
'name': 'test_drop_duplicated_metadata',
'version': '0.0.1'
},
})

self.storage.content_metadata_add([metadata_v2])

# then
actual_metadata = list(self.storage.content_metadata_get(
[self.sha1_2]))

# metadata did not change as the v2 was dropped.
self.assertEqual(actual_metadata, expected_metadata_v1)

def test_content_metadata_add_update_in_place_duplicate(self):
# given
tool_id = self.tools['swh-metadata-translator']['id']

metadata_v1 = {
'id': self.sha1_2,
'translated_metadata': {
'other': {},
'name': 'test_metadata',
'version': '0.0.1'
},		},
'indexer_configuration_id': tool_id,		example_data2={
}

# given
self.storage.content_metadata_add([metadata_v1])

# when
actual_metadata = list(self.storage.content_metadata_get(
[self.sha1_2]))

# then
expected_metadata_v1 = [{
'id': self.sha1_2,
'translated_metadata': {		'translated_metadata': {
'other': {},		'other': {},
'name': 'test_metadata',		'name': 'test_metadata',
'version': '0.0.1'		'version': '0.0.1'
},		},
'tool': self.tools['swh-metadata-translator']
}]
self.assertEqual(actual_metadata, expected_metadata_v1)

# given
metadata_v2 = metadata_v1.copy()
metadata_v2.update({
'translated_metadata': {
'other': {},
'name': 'test_update_duplicated_metadata',
'version': '0.0.1'
},
})
self.storage.content_metadata_add([metadata_v2], conflict_update=True)

actual_metadata = list(self.storage.content_metadata_get(
[self.sha1_2]))

# language did not change as the v2 was dropped.
expected_metadata_v2 = [{
'id': self.sha1_2,
'translated_metadata': {
'other': {},
'name': 'test_update_duplicated_metadata',
'version': '0.0.1'
},
'tool': self.tools['swh-metadata-translator']
}]

# metadata did change as the v2 was used to overwrite v1
self.assertEqual(actual_metadata, expected_metadata_v2)

def test_revision_metadata_missing(self):
# given
tool_id = self.tools['swh-metadata-detector']['id']

metadata = [
{
'id': self.revision_id_1,
'indexer_configuration_id': tool_id,
},
{
'id': self.revision_id_2,
'indexer_configuration_id': tool_id,
}
]

# when
actual_missing = list(self.storage.revision_metadata_missing(
metadata))

# then
self.assertEqual(list(actual_missing), [
self.revision_id_1,
self.revision_id_2,
])

# given
self.storage.revision_metadata_add([{
'id': self.revision_id_1,
'translated_metadata': {
'developmentStatus': None,
'version': None,
'operatingSystem': None,
'description': None,
'keywords': None,
'issueTracker': None,
'name': None,
'author': None,
'relatedLink': None,
'url': None,
'license': None,
'maintainer': None,
'email': None,
'softwareRequirements': None,
'identifier': None
},
'mappings': [],
'indexer_configuration_id': tool_id
}])

# when
actual_missing = list(self.storage.revision_metadata_missing(
metadata))

# then
self.assertEqual(actual_missing, [self.revision_id_2])

def test_revision_metadata_get(self):
# given
tool_id = self.tools['swh-metadata-detector']['id']

metadata_rev = {
'id': self.revision_id_2,
'translated_metadata': {
'developmentStatus': None,
'version': None,
'operatingSystem': None,
'description': None,
'keywords': None,
'issueTracker': None,
'name': None,
'author': None,
'relatedLink': None,
'url': None,
'license': None,
'maintainer': None,
'email': None,
'softwareRequirements': None,
'identifier': None
},
'mappings': ['mapping1', 'mapping2'],
'indexer_configuration_id': tool_id
}

# when
self.storage.revision_metadata_add([metadata_rev])

# then
actual_metadata = list(self.storage.revision_metadata_get(
[self.revision_id_2, self.revision_id_1]))

expected_metadata = [{
'id': self.revision_id_2,
'translated_metadata': metadata_rev['translated_metadata'],
'mappings': ['mapping1', 'mapping2'],
'tool': self.tools['swh-metadata-detector']
}]

self.maxDiff = None
self.assertEqual(actual_metadata, expected_metadata)

def test_revision_metadata_add_drop_duplicate(self):
# given
tool_id = self.tools['swh-metadata-detector']['id']

metadata_v1 = {
'id': self.revision_id_1,
'translated_metadata': {
'developmentStatus': None,
'version': None,
'operatingSystem': None,
'description': None,
'keywords': None,
'issueTracker': None,
'name': None,
'author': None,
'relatedLink': None,
'url': None,
'license': None,
'maintainer': None,
'email': None,
'softwareRequirements': None,
'identifier': None
},
'mappings': ['mapping1'],
'indexer_configuration_id': tool_id,
}

# given
self.storage.revision_metadata_add([metadata_v1])

# when
actual_metadata = list(self.storage.revision_metadata_get(
[self.revision_id_1]))

expected_metadata_v1 = [{
'id': self.revision_id_1,
'translated_metadata': metadata_v1['translated_metadata'],
'mappings': ['mapping1'],
'tool': self.tools['swh-metadata-detector']
}]

self.assertEqual(actual_metadata, expected_metadata_v1)

# given
metadata_v2 = metadata_v1.copy()
metadata_v2.update({
'translated_metadata': {
'name': 'test_metadata',
'author': 'MG',
},
'mappings': ['mapping2'],		'mappings': ['mapping2'],
})

self.storage.revision_metadata_add([metadata_v2])

# then
actual_metadata = list(self.storage.revision_metadata_get(
[self.revision_id_1]))

# metadata did not change as the v2 was dropped.
self.assertEqual(actual_metadata, expected_metadata_v1)

def test_revision_metadata_add_update_in_place_duplicate(self):
# given
tool_id = self.tools['swh-metadata-detector']['id']

metadata_v1 = {
'id': self.revision_id_2,
'translated_metadata': {
'developmentStatus': None,
'version': None,
'operatingSystem': None,
'description': None,
'keywords': None,
'issueTracker': None,
'name': None,
'author': None,
'relatedLink': None,
'url': None,
'license': None,
'maintainer': None,
'email': None,
'softwareRequirements': None,
'identifier': None
},
'mappings': ['mapping1'],
'indexer_configuration_id': tool_id,
}

# given
self.storage.revision_metadata_add([metadata_v1])

# when
actual_metadata = list(self.storage.revision_metadata_get(
[self.revision_id_2]))

# then
expected_metadata_v1 = [{
'id': self.revision_id_2,
'translated_metadata': metadata_v1['translated_metadata'],
'mappings': ['mapping1'],
'tool': self.tools['swh-metadata-detector']
}]
self.assertEqual(actual_metadata, expected_metadata_v1)

# given
metadata_v2 = metadata_v1.copy()
metadata_v2.update({
'translated_metadata': {
'name': 'test_update_duplicated_metadata',
'author': 'MG'
},		},
'mappings': ['mapping2'],		)
})
self.storage.revision_metadata_add([metadata_v2], conflict_update=True)

actual_metadata = list(self.storage.revision_metadata_get(
[self.revision_id_2]))

expected_metadata_v2 = [{
'id': self.revision_id_2,
'translated_metadata': metadata_v2['translated_metadata'],
'mappings': ['mapping2'],
'tool': self.tools['swh-metadata-detector']
}]

# metadata did change as the v2 was used to overwrite v1
self.assertEqual(actual_metadata, expected_metadata_v2)

def test_origin_intrinsic_metadata_get(self):		def test_origin_intrinsic_metadata_get(self):
# given		# given
tool_id = self.tools['swh-metadata-detector']['id']		tool_id = self.tools['swh-metadata-detector']['id']

metadata = {		metadata = {
'developmentStatus': None,		'developmentStatus': None,
'version': None,		'version': None,
▲ Show 20 Lines • Show All 668 Lines • Show Last 20 Lines