Changeset View
Changeset View
Standalone View
Standalone View
swh/indexer/tests/storage/test_storage.py
| Show First 20 Lines • Show All 1,289 Lines • ▼ Show 20 Lines | def test_revision_metadata_add_drop_duplicate(self): | ||||
| 'relatedLink': None, | 'relatedLink': None, | ||||
| 'url': None, | 'url': None, | ||||
| 'license': None, | 'license': None, | ||||
| 'maintainer': None, | 'maintainer': None, | ||||
| 'email': None, | 'email': None, | ||||
| 'softwareRequirements': None, | 'softwareRequirements': None, | ||||
| 'identifier': None | 'identifier': None | ||||
| }, | }, | ||||
| 'mappings': [], | 'mappings': ['mapping1'], | ||||
| 'indexer_configuration_id': tool_id, | 'indexer_configuration_id': tool_id, | ||||
| } | } | ||||
| # given | # given | ||||
| self.storage.revision_metadata_add([metadata_v1]) | self.storage.revision_metadata_add([metadata_v1]) | ||||
| # when | # when | ||||
| actual_metadata = list(self.storage.revision_metadata_get( | actual_metadata = list(self.storage.revision_metadata_get( | ||||
| [self.revision_id_1])) | [self.revision_id_1])) | ||||
| expected_metadata_v1 = [{ | expected_metadata_v1 = [{ | ||||
| 'id': self.revision_id_1, | 'id': self.revision_id_1, | ||||
| 'translated_metadata': metadata_v1['translated_metadata'], | 'translated_metadata': metadata_v1['translated_metadata'], | ||||
| 'mappings': [], | 'mappings': ['mapping1'], | ||||
| 'tool': self.tools['swh-metadata-detector'] | 'tool': self.tools['swh-metadata-detector'] | ||||
| }] | }] | ||||
| self.assertEqual(actual_metadata, expected_metadata_v1) | self.assertEqual(actual_metadata, expected_metadata_v1) | ||||
| # given | # given | ||||
| metadata_v2 = metadata_v1.copy() | metadata_v2 = metadata_v1.copy() | ||||
| metadata_v2.update({ | metadata_v2.update({ | ||||
| 'translated_metadata': { | 'translated_metadata': { | ||||
| 'name': 'test_metadata', | 'name': 'test_metadata', | ||||
| 'author': 'MG', | 'author': 'MG', | ||||
| }, | }, | ||||
| 'mappings': ['mapping2'], | |||||
| }) | }) | ||||
| self.storage.revision_metadata_add([metadata_v2]) | self.storage.revision_metadata_add([metadata_v2]) | ||||
| # then | # then | ||||
| actual_metadata = list(self.storage.revision_metadata_get( | actual_metadata = list(self.storage.revision_metadata_get( | ||||
| [self.revision_id_1])) | [self.revision_id_1])) | ||||
| Show All 18 Lines | def test_revision_metadata_add_update_in_place_duplicate(self): | ||||
| 'relatedLink': None, | 'relatedLink': None, | ||||
| 'url': None, | 'url': None, | ||||
| 'license': None, | 'license': None, | ||||
| 'maintainer': None, | 'maintainer': None, | ||||
| 'email': None, | 'email': None, | ||||
| 'softwareRequirements': None, | 'softwareRequirements': None, | ||||
| 'identifier': None | 'identifier': None | ||||
| }, | }, | ||||
| 'mappings': [], | 'mappings': ['mapping1'], | ||||
| 'indexer_configuration_id': tool_id, | 'indexer_configuration_id': tool_id, | ||||
| } | } | ||||
| # given | # given | ||||
| self.storage.revision_metadata_add([metadata_v1]) | self.storage.revision_metadata_add([metadata_v1]) | ||||
| # when | # when | ||||
| actual_metadata = list(self.storage.revision_metadata_get( | actual_metadata = list(self.storage.revision_metadata_get( | ||||
| [self.revision_id_2])) | [self.revision_id_2])) | ||||
| # then | # then | ||||
| expected_metadata_v1 = [{ | expected_metadata_v1 = [{ | ||||
| 'id': self.revision_id_2, | 'id': self.revision_id_2, | ||||
| 'translated_metadata': metadata_v1['translated_metadata'], | 'translated_metadata': metadata_v1['translated_metadata'], | ||||
| 'mappings': [], | 'mappings': ['mapping1'], | ||||
| 'tool': self.tools['swh-metadata-detector'] | 'tool': self.tools['swh-metadata-detector'] | ||||
| }] | }] | ||||
| self.assertEqual(actual_metadata, expected_metadata_v1) | self.assertEqual(actual_metadata, expected_metadata_v1) | ||||
| # given | # given | ||||
| metadata_v2 = metadata_v1.copy() | metadata_v2 = metadata_v1.copy() | ||||
| metadata_v2.update({ | metadata_v2.update({ | ||||
| 'translated_metadata': { | 'translated_metadata': { | ||||
| 'name': 'test_update_duplicated_metadata', | 'name': 'test_update_duplicated_metadata', | ||||
| 'author': 'MG' | 'author': 'MG' | ||||
| }, | }, | ||||
| 'mappings': ['mapping2'], | |||||
| }) | }) | ||||
| self.storage.revision_metadata_add([metadata_v2], conflict_update=True) | self.storage.revision_metadata_add([metadata_v2], conflict_update=True) | ||||
| actual_metadata = list(self.storage.revision_metadata_get( | actual_metadata = list(self.storage.revision_metadata_get( | ||||
| [self.revision_id_2])) | [self.revision_id_2])) | ||||
| expected_metadata_v2 = [{ | expected_metadata_v2 = [{ | ||||
| 'id': self.revision_id_2, | 'id': self.revision_id_2, | ||||
| 'translated_metadata': metadata_v2['translated_metadata'], | 'translated_metadata': metadata_v2['translated_metadata'], | ||||
| 'mappings': [], | 'mappings': ['mapping2'], | ||||
| 'tool': self.tools['swh-metadata-detector'] | 'tool': self.tools['swh-metadata-detector'] | ||||
| }] | }] | ||||
| # metadata did change as the v2 was used to overwrite v1 | # metadata did change as the v2 was used to overwrite v1 | ||||
| self.assertEqual(actual_metadata, expected_metadata_v2) | self.assertEqual(actual_metadata, expected_metadata_v2) | ||||
| def test_origin_intrinsic_metadata_get(self): | def test_origin_intrinsic_metadata_get(self): | ||||
| # given | # given | ||||
| ▲ Show 20 Lines • Show All 673 Lines • Show Last 20 Lines | |||||