Changeset View
Changeset View
Standalone 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.revision_id_3 = hash_to_bytes( | |||||
'7026b7c1a2af56521e9587659012345678904320') | |||||
self.origin_id_1 = 54974445 | |||||
self.origin_id_2 = 44434342 | |||||
self.origin_id_3 = 44434341 | |||||
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): | return ( | ||||
# given | missing, | ||||
tool_id = self.tools['pygments']['id'] | add__drop_duplicate, | ||||
add__update_in_place_duplicate, | |||||
languages = [ | get, | ||||
{ | ) | ||||
'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 | |||||
self.storage.content_language_add([{ | |||||
'id': self.sha1_2, | |||||
'lang': 'haskell', | |||||
'indexer_configuration_id': tool_id, | |||||
}]) | |||||
# 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.revision_id_3 = hash_to_bytes( | |||||
# then | '7026b7c1a2af56521e9587659012345678904320') | ||||
expected_languages_v1 = [{ | self.origin_id_1 = 54974445 | ||||
'id': self.sha1_2, | self.origin_id_2 = 44434342 | ||||
'lang': 'emacslisp', | self.origin_id_3 = 44434341 | ||||
'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', | |||||
ardumont: It seems we lose coverage here.
Do not migrate those tests to the new way and avoid losing… | |||||
Done Inline Actions
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 745 Lines • Show Last 20 Lines |
It seems we lose coverage here.
Do not migrate those tests to the new way and avoid losing coverage.