Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 935 Lines • ▼ Show 20 Lines | def test_origin_get_legacy(self, swh_storage): | ||||
| assert actual_origin0['id'] == id | assert actual_origin0['id'] == id | ||||
| assert actual_origin0['url'] == data.origin['url'] | assert actual_origin0['url'] == data.origin['url'] | ||||
| # lookup per id (returns dict) | # lookup per id (returns dict) | ||||
| if self._test_origin_ids: | if self._test_origin_ids: | ||||
| actual_origin1 = swh_storage.origin_get({'id': id}) | actual_origin1 = swh_storage.origin_get({'id': id}) | ||||
| assert actual_origin1 == {'id': id, | assert actual_origin1 == {'id': id, | ||||
| 'type': data.origin['type'], | |||||
| 'url': data.origin['url']} | 'url': data.origin['url']} | ||||
| def test_origin_get(self, swh_storage): | def test_origin_get(self, swh_storage): | ||||
| assert swh_storage.origin_get(data.origin) is None | assert swh_storage.origin_get(data.origin) is None | ||||
| origin_id = swh_storage.origin_add_one(data.origin) | origin_id = swh_storage.origin_add_one(data.origin) | ||||
| # lookup per url (returns id) | # lookup per url (returns id) | ||||
| actual_origin0 = swh_storage.origin_get( | actual_origin0 = swh_storage.origin_get( | ||||
| [{'url': data.origin['url']}]) | [{'url': data.origin['url']}]) | ||||
| assert len(actual_origin0) == 1 | assert len(actual_origin0) == 1 | ||||
| assert actual_origin0[0]['url'] == data.origin['url'] | assert actual_origin0[0]['url'] == data.origin['url'] | ||||
| if self._test_origin_ids: | if self._test_origin_ids: | ||||
| # lookup per id (returns dict) | # lookup per id (returns dict) | ||||
| actual_origin1 = swh_storage.origin_get([{'id': origin_id}]) | actual_origin1 = swh_storage.origin_get([{'id': origin_id}]) | ||||
| assert len(actual_origin1) == 1 | assert len(actual_origin1) == 1 | ||||
| assert actual_origin1[0] == {'id': origin_id, | assert actual_origin1[0] == {'id': origin_id, | ||||
| 'type': data.origin['type'], | |||||
| 'url': data.origin['url']} | 'url': data.origin['url']} | ||||
| def test_origin_get_consistency(self, swh_storage): | def test_origin_get_consistency(self, swh_storage): | ||||
| assert swh_storage.origin_get(data.origin) is None | assert swh_storage.origin_get(data.origin) is None | ||||
| id = swh_storage.origin_add_one(data.origin) | id = swh_storage.origin_add_one(data.origin) | ||||
| with pytest.raises(ValueError): | with pytest.raises(ValueError): | ||||
| swh_storage.origin_get([ | swh_storage.origin_get([ | ||||
| {'url': data.origin['url']}, | {'url': data.origin['url']}, | ||||
| {'id': id}]) | {'id': id}]) | ||||
| def test_origin_search_single_result(self, swh_storage): | def test_origin_search_single_result(self, swh_storage): | ||||
| found_origins = list(swh_storage.origin_search(data.origin['url'])) | found_origins = list(swh_storage.origin_search(data.origin['url'])) | ||||
| assert len(found_origins) == 0 | assert len(found_origins) == 0 | ||||
| found_origins = list(swh_storage.origin_search(data.origin['url'], | found_origins = list(swh_storage.origin_search(data.origin['url'], | ||||
| regexp=True)) | regexp=True)) | ||||
| assert len(found_origins) == 0 | assert len(found_origins) == 0 | ||||
| swh_storage.origin_add_one(data.origin) | swh_storage.origin_add_one(data.origin) | ||||
| origin_data = { | origin_data = { | ||||
| 'type': data.origin['type'], | |||||
| 'url': data.origin['url']} | 'url': data.origin['url']} | ||||
| found_origins = list(swh_storage.origin_search(data.origin['url'])) | found_origins = list(swh_storage.origin_search(data.origin['url'])) | ||||
| assert len(found_origins) == 1 | assert len(found_origins) == 1 | ||||
| if 'id' in found_origins[0]: | if 'id' in found_origins[0]: | ||||
| del found_origins[0]['id'] | del found_origins[0]['id'] | ||||
| assert found_origins[0] == origin_data | assert found_origins[0] == origin_data | ||||
| found_origins = list(swh_storage.origin_search( | found_origins = list(swh_storage.origin_search( | ||||
| '.' + data.origin['url'][1:-1] + '.', regexp=True)) | '.' + data.origin['url'][1:-1] + '.', regexp=True)) | ||||
| assert len(found_origins) == 1 | assert len(found_origins) == 1 | ||||
| if 'id' in found_origins[0]: | if 'id' in found_origins[0]: | ||||
| del found_origins[0]['id'] | del found_origins[0]['id'] | ||||
| assert found_origins[0] == origin_data | assert found_origins[0] == origin_data | ||||
| swh_storage.origin_add_one(data.origin2) | swh_storage.origin_add_one(data.origin2) | ||||
| origin2_data = { | origin2_data = {'url': data.origin2['url']} | ||||
| 'type': data.origin2['type'], | |||||
| 'url': data.origin2['url']} | |||||
| found_origins = list(swh_storage.origin_search(data.origin2['url'])) | found_origins = list(swh_storage.origin_search(data.origin2['url'])) | ||||
| assert len(found_origins) == 1 | assert len(found_origins) == 1 | ||||
| if 'id' in found_origins[0]: | if 'id' in found_origins[0]: | ||||
| del found_origins[0]['id'] | del found_origins[0]['id'] | ||||
| assert found_origins[0] == origin2_data | assert found_origins[0] == origin2_data | ||||
| found_origins = list(swh_storage.origin_search( | found_origins = list(swh_storage.origin_search( | ||||
| '.' + data.origin2['url'][1:-1] + '.', regexp=True)) | '.' + data.origin2['url'][1:-1] + '.', regexp=True)) | ||||
| ▲ Show 20 Lines • Show All 2,505 Lines • Show Last 20 Lines | |||||