Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 329 Lines • ▼ Show 20 Lines | def setUp(self): | ||||
| 'type': 'git', | 'type': 'git', | ||||
| } | } | ||||
| self.origin2 = { | self.origin2 = { | ||||
| 'url': 'file:///dev/zero', | 'url': 'file:///dev/zero', | ||||
| 'type': 'git', | 'type': 'git', | ||||
| } | } | ||||
| self.origin_metadata = { | |||||
| 'origin': self.origin, | |||||
| 'date': datetime.datetime(2015, 1, 1, 23, 0, 0, | |||||
| tzinfo=datetime.timezone.utc), | |||||
| 'provenance': 'deposit-hal', | |||||
| 'metadata': { | |||||
| 'name': 'test_origin_metadata', | |||||
| 'version': '0.0.1' | |||||
| } | |||||
| } | |||||
| self.origin_metadata2 = { | |||||
| 'origin': self.origin, | |||||
| 'date': datetime.datetime(2017, 1, 1, 23, 0, 0, | |||||
| tzinfo=datetime.timezone.utc), | |||||
| 'provenance': 'lister-github', | |||||
| 'metadata': { | |||||
| 'name': 'test_origin_metadata', | |||||
| 'version': '0.0.1' | |||||
| } | |||||
| } | |||||
| self.date_visit1 = datetime.datetime(2015, 1, 1, 23, 0, 0, | self.date_visit1 = datetime.datetime(2015, 1, 1, 23, 0, 0, | ||||
| tzinfo=datetime.timezone.utc) | tzinfo=datetime.timezone.utc) | ||||
| self.occurrence = { | self.occurrence = { | ||||
| 'branch': b'master', | 'branch': b'master', | ||||
| 'target': b'67890123456789012345', | 'target': b'67890123456789012345', | ||||
| 'target_type': 'revision', | 'target_type': 'revision', | ||||
| } | } | ||||
| ▲ Show 20 Lines • Show All 3,078 Lines • ▼ Show 20 Lines | def indexer_configuration_metadata_get(self): | ||||
| actual_tool = self.storage.indexer_configuration_get(tool) | actual_tool = self.storage.indexer_configuration_get(tool) | ||||
| expected_tool = tool.copy() | expected_tool = tool.copy() | ||||
| expected_tool['id'] = actual_tool['id'] | expected_tool['id'] = actual_tool['id'] | ||||
| self.assertEqual(expected_tool, actual_tool) | self.assertEqual(expected_tool, actual_tool) | ||||
| @istest | |||||
| def origin_metadata_add(self): | |||||
| # given | |||||
| origin_metadata0 = self.storage.origin_metadata_get(self.origin) | |||||
| self.assertIsNone(origin_metadata0) | |||||
| origin_id = self.storage.origin_add([self.origin]) | |||||
| # when adding for the same origin 2 metadatas | |||||
| o_m_id1 = self.storage.origin_metadata_add(self.origin_metadata) | |||||
| o_m_id2 = self.storage.origin_metadata_add(self.origin_metadata2) | |||||
| actual_origin_metadata1 = self.storage.origin_metadata_get(o_m_id1) | |||||
| actual_origin_metadata2 = self.storage.origin_metadata_get(o_m_id2) | |||||
| # then | |||||
| self.assertEqual(actual_origin_metadata1['id'], o_m_id1) | |||||
| self.assertEqual(actual_origin_metadata2['id'], o_m_id2) | |||||
| self.assertEqual(actual_origin_metadata1['origin_id'], origin_id) | |||||
| self.assertEqual(actual_origin_metadata2['origin_id'], origin_id) | |||||
| @istest | |||||
| def origin_metadata_get(self): | |||||
| # given | |||||
| origin_id = self.storage.origin_add([self.origin]) | |||||
| # when adding for the same origin 2 metadatas | |||||
| o_m_id1 = self.storage.origin_metadata_add(self.origin_metadata) | |||||
| o_m_id2 = self.storage.origin_metadata_add(self.origin_metadata2) | |||||
| all_metadatas = self.storage.origin_metadata_get_all({ | |||||
| 'origin_id': origin_id, | |||||
| }) | |||||
| m_by_provenance = self.storage.origin_metadata_get_by_provenance({ | |||||
| 'origin_id': origin_id, | |||||
| 'provenance': self.origin_metadata['provenance'] | |||||
| }) | |||||
| expected_results = [{ | |||||
| 'id': o_m_id1, | |||||
| 'origin_id': origin_id, | |||||
| 'date': datetime.datetime(2015, 1, 1, 23, 0, 0, | |||||
| tzinfo=datetime.timezone.utc), | |||||
| 'provenance': 'deposit-hal', | |||||
| 'metadata': { | |||||
| 'name': 'test_origin_metadata', | |||||
| 'version': '0.0.1' | |||||
| } | |||||
| }, { | |||||
| 'id': o_m_id2, | |||||
| 'origin_id': origin_id, | |||||
| 'date': datetime.datetime(2017, 1, 1, 23, 0, 0, | |||||
| tzinfo=datetime.timezone.utc), | |||||
| 'provenance': 'lister-hal', | |||||
| 'metadata': { | |||||
| 'name': 'test_origin_metadata', | |||||
| 'version': '0.0.1' | |||||
| } | |||||
| }] | |||||
| # then | |||||
| self.assertEqual(len(all_metadatas), 2) | |||||
| self.assertEqual(len(m_by_provenance), 1) | |||||
| self.assertEqual(all_metadatas, expected_results) | |||||
| class TestLocalStorage(CommonTestStorage, unittest.TestCase): | class TestLocalStorage(CommonTestStorage, unittest.TestCase): | ||||
| """Test the local storage""" | """Test the local storage""" | ||||
| # Can only be tested with local storage as you can't mock | # Can only be tested with local storage as you can't mock | ||||
| # datetimes for the remote server | # datetimes for the remote server | ||||
| @istest | @istest | ||||
| def fetch_history(self): | def fetch_history(self): | ||||
| Show All 10 Lines | def fetch_history(self): | ||||
| fetch_history = self.storage.fetch_history_get(fetch_history_id) | fetch_history = self.storage.fetch_history_get(fetch_history_id) | ||||
| expected_fetch_history = self.fetch_history_data.copy() | expected_fetch_history = self.fetch_history_data.copy() | ||||
| expected_fetch_history['id'] = fetch_history_id | expected_fetch_history['id'] = fetch_history_id | ||||
| expected_fetch_history['origin'] = origin | expected_fetch_history['origin'] = origin | ||||
| expected_fetch_history['date'] = self.fetch_history_date | expected_fetch_history['date'] = self.fetch_history_date | ||||
| expected_fetch_history['duration'] = self.fetch_history_duration | expected_fetch_history['duration'] = self.fetch_history_duration | ||||
ardumont: Maybe add another origin_metadata from another provenance.
That way, this shows the filtering. | |||||
| self.assertEqual(expected_fetch_history, fetch_history) | self.assertEqual(expected_fetch_history, fetch_history) | ||||
| # The remote API doesn't expose _person_add | # The remote API doesn't expose _person_add | ||||
| @istest | @istest | ||||
| def person_get(self): | def person_get(self): | ||||
| # given | # given | ||||
| person0 = { | person0 = { | ||||
| 'fullname': b'bob <alice@bob>', | 'fullname': b'bob <alice@bob>', | ||||
| Show All 34 Lines | class AlteringSchemaTest(BaseTestStorage, unittest.TestCase): | ||||
| """This class is dedicated for the rare case where the schema needs to | """This class is dedicated for the rare case where the schema needs to | ||||
| be altered dynamically. | be altered dynamically. | ||||
| Otherwise, the tests could be blocking when ran altogether. | Otherwise, the tests could be blocking when ran altogether. | ||||
| """ | """ | ||||
| @istest | @istest | ||||
| def content_update(self): | def content_update(self): | ||||
| cont = copy.deepcopy(self.cont) | cont = copy.deepcopy(self.cont) | ||||
Done Inline ActionsThis tends to agree with my previous assertions :) ardumont: This tends to agree with my previous assertions :) | |||||
| self.storage.content_add([cont]) | self.storage.content_add([cont]) | ||||
| # alter the sha1_git for example | # alter the sha1_git for example | ||||
| cont['sha1_git'] = hash_to_bytes( | cont['sha1_git'] = hash_to_bytes( | ||||
| '3a60a5275d0333bf13468e8b3dcab90f4046e654') | '3a60a5275d0333bf13468e8b3dcab90f4046e654') | ||||
| self.storage.content_update([cont], keys=['sha1_git']) | self.storage.content_update([cont], keys=['sha1_git']) | ||||
| Show All 37 Lines | |||||
Maybe add another origin_metadata from another provenance.
That way, this shows the filtering.