Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 3,221 Lines • ▼ Show 20 Lines | def test_content_metadata_add_duplicate(self, swh_storage): | ||||
| swh_storage.metadata_authority_add(**authority) | swh_storage.metadata_authority_add(**authority) | ||||
| swh_storage.content_metadata_add(**data.content_metadata) | swh_storage.content_metadata_add(**data.content_metadata) | ||||
| swh_storage.content_metadata_add(**data.content_metadata2) | swh_storage.content_metadata_add(**data.content_metadata2) | ||||
| swh_storage.content_metadata_add(**new_content_metadata2) | swh_storage.content_metadata_add(**new_content_metadata2) | ||||
| result = swh_storage.content_metadata_get(content_swhid, authority) | result = swh_storage.content_metadata_get(content_swhid, authority) | ||||
| assert result["next_page_token"] is None | assert result["next_page_token"] is None | ||||
| assert [data.content_metadata, new_content_metadata2] == list( | |||||
| sorted(result["results"], key=lambda x: x["discovery_date"],) | expected_results1 = (data.content_metadata, new_content_metadata2) | ||||
| expected_results2 = (data.content_metadata, data.content_metadata2) | |||||
| assert tuple(sorted(result["results"], key=lambda x: x["discovery_date"],)) in ( | |||||
| expected_results1, # cassandra | |||||
| expected_results2, # postgresql | |||||
| ) | ) | ||||
| def test_content_metadata_add_dict(self, swh_storage): | def test_content_metadata_add_dict(self, swh_storage): | ||||
| fetcher = data.metadata_fetcher | fetcher = data.metadata_fetcher | ||||
| authority = data.metadata_authority | authority = data.metadata_authority | ||||
| swh_storage.metadata_fetcher_add(**fetcher) | swh_storage.metadata_fetcher_add(**fetcher) | ||||
| swh_storage.metadata_authority_add(**authority) | swh_storage.metadata_authority_add(**authority) | ||||
| ▲ Show 20 Lines • Show All 228 Lines • ▼ Show 20 Lines | def test_origin_metadata_add_duplicate(self, swh_storage): | ||||
| swh_storage.metadata_authority_add(**authority) | swh_storage.metadata_authority_add(**authority) | ||||
| swh_storage.origin_metadata_add(**data.origin_metadata) | swh_storage.origin_metadata_add(**data.origin_metadata) | ||||
| swh_storage.origin_metadata_add(**data.origin_metadata2) | swh_storage.origin_metadata_add(**data.origin_metadata2) | ||||
| swh_storage.origin_metadata_add(**new_origin_metadata2) | swh_storage.origin_metadata_add(**new_origin_metadata2) | ||||
| result = swh_storage.origin_metadata_get(origin["url"], authority) | result = swh_storage.origin_metadata_get(origin["url"], authority) | ||||
| assert result["next_page_token"] is None | assert result["next_page_token"] is None | ||||
| assert [data.origin_metadata, new_origin_metadata2] == list( | |||||
| sorted(result["results"], key=lambda x: x["discovery_date"],) | # which of the two behavior happens is backend-specific. | ||||
| expected_results1 = (data.origin_metadata, new_origin_metadata2) | |||||
| expected_results2 = (data.origin_metadata, data.origin_metadata2) | |||||
| assert tuple(sorted(result["results"], key=lambda x: x["discovery_date"],)) in ( | |||||
ardumont: maybe add something to tell this one is pg, this one is cass or something?
If i'm not mistaken… | |||||
| expected_results1, # cassandra | |||||
| expected_results2, # postgresql | |||||
| ) | ) | ||||
| def test_origin_metadata_add_dict(self, swh_storage): | def test_origin_metadata_add_dict(self, swh_storage): | ||||
| origin = data.origin | origin = data.origin | ||||
| fetcher = data.metadata_fetcher | fetcher = data.metadata_fetcher | ||||
| authority = data.metadata_authority | authority = data.metadata_authority | ||||
| assert swh_storage.origin_add([origin]) == {"origin:add": 1} | assert swh_storage.origin_add([origin]) == {"origin:add": 1} | ||||
| ▲ Show 20 Lines • Show All 669 Lines • Show Last 20 Lines | |||||
maybe add something to tell this one is pg, this one is cass or something?
If i'm not mistaken, that'd be: