Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 3,344 Lines • ▼ Show 20 Lines | def test_content_metadata_add(self, swh_storage, sample_data): | ||||
| swh_storage.metadata_fetcher_add([fetcher]) | swh_storage.metadata_fetcher_add([fetcher]) | ||||
| swh_storage.metadata_authority_add([authority]) | swh_storage.metadata_authority_add([authority]) | ||||
| swh_storage.raw_extrinsic_metadata_add(content_metadata) | swh_storage.raw_extrinsic_metadata_add(content_metadata) | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.CONTENT, content_swhid, authority | MetadataTargetType.CONTENT, content_swhid, authority | ||||
| ) | ) | ||||
| assert result["next_page_token"] is None | assert result.next_page_token is None | ||||
| assert list(sorted(result["results"], key=lambda x: x.discovery_date,)) == list( | assert list(sorted(result.results, key=lambda x: x.discovery_date,)) == list( | ||||
| content_metadata | content_metadata | ||||
| ) | ) | ||||
| actual_objects = list(swh_storage.journal_writer.journal.objects) | actual_objects = list(swh_storage.journal_writer.journal.objects) | ||||
| expected_objects = [ | expected_objects = [ | ||||
| ("metadata_authority", authority), | ("metadata_authority", authority), | ||||
| ("metadata_fetcher", fetcher), | ("metadata_fetcher", fetcher), | ||||
| ] + [("raw_extrinsic_metadata", item) for item in content_metadata] | ] + [("raw_extrinsic_metadata", item) for item in content_metadata] | ||||
| Show All 19 Lines | def test_content_metadata_add_duplicate(self, swh_storage, sample_data): | ||||
| swh_storage.metadata_authority_add([authority]) | swh_storage.metadata_authority_add([authority]) | ||||
| swh_storage.raw_extrinsic_metadata_add([content_metadata, content_metadata2]) | swh_storage.raw_extrinsic_metadata_add([content_metadata, content_metadata2]) | ||||
| swh_storage.raw_extrinsic_metadata_add([new_content_metadata2]) | swh_storage.raw_extrinsic_metadata_add([new_content_metadata2]) | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.CONTENT, content_swhid, authority | MetadataTargetType.CONTENT, content_swhid, authority | ||||
| ) | ) | ||||
| assert result["next_page_token"] is None | assert result.next_page_token is None | ||||
| expected_results1 = (content_metadata, new_content_metadata2) | expected_results1 = (content_metadata, new_content_metadata2) | ||||
| expected_results2 = (content_metadata, content_metadata2) | expected_results2 = (content_metadata, content_metadata2) | ||||
| assert tuple(sorted(result["results"], key=lambda x: x.discovery_date,)) in ( | assert tuple(sorted(result.results, key=lambda x: x.discovery_date,)) in ( | ||||
| expected_results1, # cassandra | expected_results1, # cassandra | ||||
| expected_results2, # postgresql | expected_results2, # postgresql | ||||
| ) | ) | ||||
| def test_content_metadata_get(self, swh_storage, sample_data): | def test_content_metadata_get(self, swh_storage, sample_data): | ||||
| content, content2 = sample_data.contents[:2] | content, content2 = sample_data.contents[:2] | ||||
| fetcher, fetcher2 = sample_data.fetchers[:2] | fetcher, fetcher2 = sample_data.fetchers[:2] | ||||
| authority, authority2 = sample_data.authorities[:2] | authority, authority2 = sample_data.authorities[:2] | ||||
| Show All 17 Lines | def test_content_metadata_get(self, swh_storage, sample_data): | ||||
| content1_metadata3, | content1_metadata3, | ||||
| content2_metadata, | content2_metadata, | ||||
| ] | ] | ||||
| ) | ) | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.CONTENT, content1_swhid, authority | MetadataTargetType.CONTENT, content1_swhid, authority | ||||
| ) | ) | ||||
| assert result["next_page_token"] is None | assert result.next_page_token is None | ||||
| assert [content1_metadata1, content1_metadata2] == list( | assert [content1_metadata1, content1_metadata2] == list( | ||||
| sorted(result["results"], key=lambda x: x.discovery_date,) | sorted(result.results, key=lambda x: x.discovery_date,) | ||||
| ) | ) | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.CONTENT, content1_swhid, authority2 | MetadataTargetType.CONTENT, content1_swhid, authority2 | ||||
| ) | ) | ||||
| assert result["next_page_token"] is None | assert result.next_page_token is None | ||||
| assert [content1_metadata3] == list( | assert [content1_metadata3] == list( | ||||
| sorted(result["results"], key=lambda x: x.discovery_date,) | sorted(result.results, key=lambda x: x.discovery_date,) | ||||
| ) | ) | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.CONTENT, content2_swhid, authority | MetadataTargetType.CONTENT, content2_swhid, authority | ||||
| ) | ) | ||||
| assert result["next_page_token"] is None | assert result.next_page_token is None | ||||
| assert [content2_metadata] == list(result["results"],) | assert [content2_metadata] == list(result.results,) | ||||
| def test_content_metadata_get_after(self, swh_storage, sample_data): | def test_content_metadata_get_after(self, swh_storage, sample_data): | ||||
| content = sample_data.content | content = sample_data.content | ||||
| fetcher = sample_data.metadata_fetcher | fetcher = sample_data.metadata_fetcher | ||||
| authority = sample_data.metadata_authority | authority = sample_data.metadata_authority | ||||
| content_metadata, content_metadata2 = sample_data.content_metadata[:2] | content_metadata, content_metadata2 = sample_data.content_metadata[:2] | ||||
| content_swhid = SWHID(object_type="content", object_id=content.sha1_git) | content_swhid = SWHID(object_type="content", object_id=content.sha1_git) | ||||
| swh_storage.metadata_fetcher_add([fetcher]) | swh_storage.metadata_fetcher_add([fetcher]) | ||||
| swh_storage.metadata_authority_add([authority]) | swh_storage.metadata_authority_add([authority]) | ||||
| swh_storage.raw_extrinsic_metadata_add([content_metadata, content_metadata2]) | swh_storage.raw_extrinsic_metadata_add([content_metadata, content_metadata2]) | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.CONTENT, | MetadataTargetType.CONTENT, | ||||
| content_swhid, | content_swhid, | ||||
| authority, | authority, | ||||
| after=content_metadata.discovery_date - timedelta(seconds=1), | after=content_metadata.discovery_date - timedelta(seconds=1), | ||||
| ) | ) | ||||
| assert result["next_page_token"] is None | assert result.next_page_token is None | ||||
| assert [content_metadata, content_metadata2] == list( | assert [content_metadata, content_metadata2] == list( | ||||
| sorted(result["results"], key=lambda x: x.discovery_date,) | sorted(result.results, key=lambda x: x.discovery_date,) | ||||
| ) | ) | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.CONTENT, | MetadataTargetType.CONTENT, | ||||
| content_swhid, | content_swhid, | ||||
| authority, | authority, | ||||
| after=content_metadata.discovery_date, | after=content_metadata.discovery_date, | ||||
| ) | ) | ||||
| assert result["next_page_token"] is None | assert result.next_page_token is None | ||||
| assert result["results"] == [content_metadata2] | assert result.results == [content_metadata2] | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.CONTENT, | MetadataTargetType.CONTENT, | ||||
| content_swhid, | content_swhid, | ||||
| authority, | authority, | ||||
| after=content_metadata2.discovery_date, | after=content_metadata2.discovery_date, | ||||
| ) | ) | ||||
| assert result["next_page_token"] is None | assert result.next_page_token is None | ||||
| assert result["results"] == [] | assert result.results == [] | ||||
| def test_content_metadata_get_paginate(self, swh_storage, sample_data): | def test_content_metadata_get_paginate(self, swh_storage, sample_data): | ||||
| content = sample_data.content | content = sample_data.content | ||||
| fetcher = sample_data.metadata_fetcher | fetcher = sample_data.metadata_fetcher | ||||
| authority = sample_data.metadata_authority | authority = sample_data.metadata_authority | ||||
| content_metadata, content_metadata2 = sample_data.content_metadata[:2] | content_metadata, content_metadata2 = sample_data.content_metadata[:2] | ||||
| content_swhid = SWHID(object_type="content", object_id=content.sha1_git) | content_swhid = SWHID(object_type="content", object_id=content.sha1_git) | ||||
| swh_storage.metadata_fetcher_add([fetcher]) | swh_storage.metadata_fetcher_add([fetcher]) | ||||
| swh_storage.metadata_authority_add([authority]) | swh_storage.metadata_authority_add([authority]) | ||||
| swh_storage.raw_extrinsic_metadata_add([content_metadata, content_metadata2]) | swh_storage.raw_extrinsic_metadata_add([content_metadata, content_metadata2]) | ||||
| swh_storage.raw_extrinsic_metadata_get( | swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.CONTENT, content_swhid, authority | MetadataTargetType.CONTENT, content_swhid, authority | ||||
| ) | ) | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.CONTENT, content_swhid, authority, limit=1 | MetadataTargetType.CONTENT, content_swhid, authority, limit=1 | ||||
| ) | ) | ||||
| assert result["next_page_token"] is not None | assert result.next_page_token is not None | ||||
| assert result["results"] == [content_metadata] | assert result.results == [content_metadata] | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.CONTENT, | MetadataTargetType.CONTENT, | ||||
| content_swhid, | content_swhid, | ||||
| authority, | authority, | ||||
| limit=1, | limit=1, | ||||
| page_token=result["next_page_token"], | page_token=result.next_page_token, | ||||
| ) | ) | ||||
| assert result["next_page_token"] is None | assert result.next_page_token is None | ||||
| assert result["results"] == [content_metadata2] | assert result.results == [content_metadata2] | ||||
| def test_content_metadata_get_paginate_same_date(self, swh_storage, sample_data): | def test_content_metadata_get_paginate_same_date(self, swh_storage, sample_data): | ||||
| content = sample_data.content | content = sample_data.content | ||||
| fetcher1, fetcher2 = sample_data.fetchers[:2] | fetcher1, fetcher2 = sample_data.fetchers[:2] | ||||
| authority = sample_data.metadata_authority | authority = sample_data.metadata_authority | ||||
| content_metadata, content_metadata2 = sample_data.content_metadata[:2] | content_metadata, content_metadata2 = sample_data.content_metadata[:2] | ||||
| content_swhid = SWHID(object_type="content", object_id=content.sha1_git) | content_swhid = SWHID(object_type="content", object_id=content.sha1_git) | ||||
| Show All 9 Lines | def test_content_metadata_get_paginate_same_date(self, swh_storage, sample_data): | ||||
| swh_storage.raw_extrinsic_metadata_add( | swh_storage.raw_extrinsic_metadata_add( | ||||
| [content_metadata, new_content_metadata2] | [content_metadata, new_content_metadata2] | ||||
| ) | ) | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.CONTENT, content_swhid, authority, limit=1 | MetadataTargetType.CONTENT, content_swhid, authority, limit=1 | ||||
| ) | ) | ||||
| assert result["next_page_token"] is not None | assert result.next_page_token is not None | ||||
| assert result["results"] == [content_metadata] | assert result.results == [content_metadata] | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.CONTENT, | MetadataTargetType.CONTENT, | ||||
| content_swhid, | content_swhid, | ||||
| authority, | authority, | ||||
| limit=1, | limit=1, | ||||
| page_token=result["next_page_token"], | page_token=result.next_page_token, | ||||
| ) | ) | ||||
| assert result["next_page_token"] is None | assert result.next_page_token is None | ||||
| assert result["results"] == [new_content_metadata2] | assert result.results == [new_content_metadata2] | ||||
| def test_content_metadata_get__invalid_id(self, swh_storage, sample_data): | def test_content_metadata_get__invalid_id(self, swh_storage, sample_data): | ||||
| origin = sample_data.origin | origin = sample_data.origin | ||||
| fetcher = sample_data.metadata_fetcher | fetcher = sample_data.metadata_fetcher | ||||
| authority = sample_data.metadata_authority | authority = sample_data.metadata_authority | ||||
| content_metadata, content_metadata2 = sample_data.content_metadata[:2] | content_metadata, content_metadata2 = sample_data.content_metadata[:2] | ||||
| swh_storage.metadata_fetcher_add([fetcher]) | swh_storage.metadata_fetcher_add([fetcher]) | ||||
| Show All 16 Lines | def test_origin_metadata_add(self, swh_storage, sample_data): | ||||
| swh_storage.metadata_fetcher_add([fetcher]) | swh_storage.metadata_fetcher_add([fetcher]) | ||||
| swh_storage.metadata_authority_add([authority]) | swh_storage.metadata_authority_add([authority]) | ||||
| swh_storage.raw_extrinsic_metadata_add([origin_metadata, origin_metadata2]) | swh_storage.raw_extrinsic_metadata_add([origin_metadata, origin_metadata2]) | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.ORIGIN, origin.url, authority | MetadataTargetType.ORIGIN, origin.url, authority | ||||
| ) | ) | ||||
| assert result["next_page_token"] is None | assert result.next_page_token is None | ||||
| assert list(sorted(result["results"], key=lambda x: x.discovery_date)) == [ | assert list(sorted(result.results, key=lambda x: x.discovery_date)) == [ | ||||
| origin_metadata, | origin_metadata, | ||||
| origin_metadata2, | origin_metadata2, | ||||
| ] | ] | ||||
| actual_objects = list(swh_storage.journal_writer.journal.objects) | actual_objects = list(swh_storage.journal_writer.journal.objects) | ||||
| expected_objects = [ | expected_objects = [ | ||||
| ("metadata_authority", authority), | ("metadata_authority", authority), | ||||
| ("metadata_fetcher", fetcher), | ("metadata_fetcher", fetcher), | ||||
| Show All 20 Lines | def test_origin_metadata_add_duplicate(self, swh_storage, sample_data): | ||||
| swh_storage.metadata_authority_add([authority]) | swh_storage.metadata_authority_add([authority]) | ||||
| swh_storage.raw_extrinsic_metadata_add([origin_metadata, origin_metadata2]) | swh_storage.raw_extrinsic_metadata_add([origin_metadata, origin_metadata2]) | ||||
| swh_storage.raw_extrinsic_metadata_add([new_origin_metadata2]) | swh_storage.raw_extrinsic_metadata_add([new_origin_metadata2]) | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.ORIGIN, origin.url, authority | MetadataTargetType.ORIGIN, origin.url, authority | ||||
| ) | ) | ||||
| assert result["next_page_token"] is None | assert result.next_page_token is None | ||||
| # which of the two behavior happens is backend-specific. | # which of the two behavior happens is backend-specific. | ||||
| expected_results1 = (origin_metadata, new_origin_metadata2) | expected_results1 = (origin_metadata, new_origin_metadata2) | ||||
| expected_results2 = (origin_metadata, origin_metadata2) | expected_results2 = (origin_metadata, origin_metadata2) | ||||
| assert tuple(sorted(result["results"], key=lambda x: x.discovery_date,)) in ( | assert tuple(sorted(result.results, key=lambda x: x.discovery_date,)) in ( | ||||
| expected_results1, # cassandra | expected_results1, # cassandra | ||||
| expected_results2, # postgresql | expected_results2, # postgresql | ||||
| ) | ) | ||||
| def test_origin_metadata_get(self, swh_storage, sample_data): | def test_origin_metadata_get(self, swh_storage, sample_data): | ||||
| origin, origin2 = sample_data.origins[:2] | origin, origin2 = sample_data.origins[:2] | ||||
| fetcher, fetcher2 = sample_data.fetchers[:2] | fetcher, fetcher2 = sample_data.fetchers[:2] | ||||
| authority, authority2 = sample_data.authorities[:2] | authority, authority2 = sample_data.authorities[:2] | ||||
| Show All 12 Lines | def test_origin_metadata_get(self, swh_storage, sample_data): | ||||
| swh_storage.raw_extrinsic_metadata_add( | swh_storage.raw_extrinsic_metadata_add( | ||||
| [origin1_metadata1, origin1_metadata2, origin1_metadata3, origin2_metadata] | [origin1_metadata1, origin1_metadata2, origin1_metadata3, origin2_metadata] | ||||
| ) | ) | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.ORIGIN, origin.url, authority | MetadataTargetType.ORIGIN, origin.url, authority | ||||
| ) | ) | ||||
| assert result["next_page_token"] is None | assert result.next_page_token is None | ||||
| assert [origin1_metadata1, origin1_metadata2] == list( | assert [origin1_metadata1, origin1_metadata2] == list( | ||||
| sorted(result["results"], key=lambda x: x.discovery_date,) | sorted(result.results, key=lambda x: x.discovery_date,) | ||||
| ) | ) | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.ORIGIN, origin.url, authority2 | MetadataTargetType.ORIGIN, origin.url, authority2 | ||||
| ) | ) | ||||
| assert result["next_page_token"] is None | assert result.next_page_token is None | ||||
| assert [origin1_metadata3] == list( | assert [origin1_metadata3] == list( | ||||
| sorted(result["results"], key=lambda x: x.discovery_date,) | sorted(result.results, key=lambda x: x.discovery_date,) | ||||
| ) | ) | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.ORIGIN, origin2.url, authority | MetadataTargetType.ORIGIN, origin2.url, authority | ||||
| ) | ) | ||||
| assert result["next_page_token"] is None | assert result.next_page_token is None | ||||
| assert [origin2_metadata] == list(result["results"],) | assert [origin2_metadata] == list(result.results,) | ||||
| def test_origin_metadata_get_after(self, swh_storage, sample_data): | def test_origin_metadata_get_after(self, swh_storage, sample_data): | ||||
| origin = sample_data.origin | origin = sample_data.origin | ||||
| fetcher = sample_data.metadata_fetcher | fetcher = sample_data.metadata_fetcher | ||||
| authority = sample_data.metadata_authority | authority = sample_data.metadata_authority | ||||
| origin_metadata, origin_metadata2 = sample_data.origin_metadata[:2] | origin_metadata, origin_metadata2 = sample_data.origin_metadata[:2] | ||||
| assert swh_storage.origin_add([origin]) == {"origin:add": 1} | assert swh_storage.origin_add([origin]) == {"origin:add": 1} | ||||
| swh_storage.metadata_fetcher_add([fetcher]) | swh_storage.metadata_fetcher_add([fetcher]) | ||||
| swh_storage.metadata_authority_add([authority]) | swh_storage.metadata_authority_add([authority]) | ||||
| swh_storage.raw_extrinsic_metadata_add([origin_metadata, origin_metadata2]) | swh_storage.raw_extrinsic_metadata_add([origin_metadata, origin_metadata2]) | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.ORIGIN, | MetadataTargetType.ORIGIN, | ||||
| origin.url, | origin.url, | ||||
| authority, | authority, | ||||
| after=origin_metadata.discovery_date - timedelta(seconds=1), | after=origin_metadata.discovery_date - timedelta(seconds=1), | ||||
| ) | ) | ||||
| assert result["next_page_token"] is None | assert result.next_page_token is None | ||||
| assert list(sorted(result["results"], key=lambda x: x.discovery_date,)) == [ | assert list(sorted(result.results, key=lambda x: x.discovery_date,)) == [ | ||||
| origin_metadata, | origin_metadata, | ||||
| origin_metadata2, | origin_metadata2, | ||||
| ] | ] | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.ORIGIN, | MetadataTargetType.ORIGIN, | ||||
| origin.url, | origin.url, | ||||
| authority, | authority, | ||||
| after=origin_metadata.discovery_date, | after=origin_metadata.discovery_date, | ||||
| ) | ) | ||||
| assert result["next_page_token"] is None | assert result.next_page_token is None | ||||
| assert result["results"] == [origin_metadata2] | assert result.results == [origin_metadata2] | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.ORIGIN, | MetadataTargetType.ORIGIN, | ||||
| origin.url, | origin.url, | ||||
| authority, | authority, | ||||
| after=origin_metadata2.discovery_date, | after=origin_metadata2.discovery_date, | ||||
| ) | ) | ||||
| assert result["next_page_token"] is None | assert result.next_page_token is None | ||||
| assert result["results"] == [] | assert result.results == [] | ||||
| def test_origin_metadata_get_paginate(self, swh_storage, sample_data): | def test_origin_metadata_get_paginate(self, swh_storage, sample_data): | ||||
| origin = sample_data.origin | origin = sample_data.origin | ||||
| fetcher = sample_data.metadata_fetcher | fetcher = sample_data.metadata_fetcher | ||||
| authority = sample_data.metadata_authority | authority = sample_data.metadata_authority | ||||
| origin_metadata, origin_metadata2 = sample_data.origin_metadata[:2] | origin_metadata, origin_metadata2 = sample_data.origin_metadata[:2] | ||||
| assert swh_storage.origin_add([origin]) == {"origin:add": 1} | assert swh_storage.origin_add([origin]) == {"origin:add": 1} | ||||
| swh_storage.metadata_fetcher_add([fetcher]) | swh_storage.metadata_fetcher_add([fetcher]) | ||||
| swh_storage.metadata_authority_add([authority]) | swh_storage.metadata_authority_add([authority]) | ||||
| swh_storage.raw_extrinsic_metadata_add([origin_metadata, origin_metadata2]) | swh_storage.raw_extrinsic_metadata_add([origin_metadata, origin_metadata2]) | ||||
| swh_storage.raw_extrinsic_metadata_get( | swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.ORIGIN, origin.url, authority | MetadataTargetType.ORIGIN, origin.url, authority | ||||
| ) | ) | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.ORIGIN, origin.url, authority, limit=1 | MetadataTargetType.ORIGIN, origin.url, authority, limit=1 | ||||
| ) | ) | ||||
| assert result["next_page_token"] is not None | assert result.next_page_token is not None | ||||
| assert result["results"] == [origin_metadata] | assert result.results == [origin_metadata] | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.ORIGIN, | MetadataTargetType.ORIGIN, | ||||
| origin.url, | origin.url, | ||||
| authority, | authority, | ||||
| limit=1, | limit=1, | ||||
| page_token=result["next_page_token"], | page_token=result.next_page_token, | ||||
| ) | ) | ||||
| assert result["next_page_token"] is None | assert result.next_page_token is None | ||||
| assert result["results"] == [origin_metadata2] | assert result.results == [origin_metadata2] | ||||
| def test_origin_metadata_get_paginate_same_date(self, swh_storage, sample_data): | def test_origin_metadata_get_paginate_same_date(self, swh_storage, sample_data): | ||||
| origin = sample_data.origin | origin = sample_data.origin | ||||
| fetcher1, fetcher2 = sample_data.fetchers[:2] | fetcher1, fetcher2 = sample_data.fetchers[:2] | ||||
| authority = sample_data.metadata_authority | authority = sample_data.metadata_authority | ||||
| origin_metadata, origin_metadata2 = sample_data.origin_metadata[:2] | origin_metadata, origin_metadata2 = sample_data.origin_metadata[:2] | ||||
| assert swh_storage.origin_add([origin]) == {"origin:add": 1} | assert swh_storage.origin_add([origin]) == {"origin:add": 1} | ||||
| swh_storage.metadata_fetcher_add([fetcher1, fetcher2]) | swh_storage.metadata_fetcher_add([fetcher1, fetcher2]) | ||||
| swh_storage.metadata_authority_add([authority]) | swh_storage.metadata_authority_add([authority]) | ||||
| new_origin_metadata2 = attr.evolve( | new_origin_metadata2 = attr.evolve( | ||||
| origin_metadata2, | origin_metadata2, | ||||
| discovery_date=origin_metadata2.discovery_date, | discovery_date=origin_metadata2.discovery_date, | ||||
| fetcher=attr.evolve(fetcher2, metadata=None), | fetcher=attr.evolve(fetcher2, metadata=None), | ||||
| ) | ) | ||||
| swh_storage.raw_extrinsic_metadata_add([origin_metadata, new_origin_metadata2]) | swh_storage.raw_extrinsic_metadata_add([origin_metadata, new_origin_metadata2]) | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.ORIGIN, origin.url, authority, limit=1 | MetadataTargetType.ORIGIN, origin.url, authority, limit=1 | ||||
| ) | ) | ||||
| assert result["next_page_token"] is not None | assert result.next_page_token is not None | ||||
| assert result["results"] == [origin_metadata] | assert result.results == [origin_metadata] | ||||
| result = swh_storage.raw_extrinsic_metadata_get( | result = swh_storage.raw_extrinsic_metadata_get( | ||||
| MetadataTargetType.ORIGIN, | MetadataTargetType.ORIGIN, | ||||
| origin.url, | origin.url, | ||||
| authority, | authority, | ||||
| limit=1, | limit=1, | ||||
| page_token=result["next_page_token"], | page_token=result.next_page_token, | ||||
| ) | ) | ||||
| assert result["next_page_token"] is None | assert result.next_page_token is None | ||||
| assert result["results"] == [new_origin_metadata2] | assert result.results == [new_origin_metadata2] | ||||
| def test_origin_metadata_add_missing_authority(self, swh_storage, sample_data): | def test_origin_metadata_add_missing_authority(self, swh_storage, sample_data): | ||||
| origin = sample_data.origin | origin = sample_data.origin | ||||
| fetcher = sample_data.metadata_fetcher | fetcher = sample_data.metadata_fetcher | ||||
| origin_metadata, origin_metadata2 = sample_data.origin_metadata[:2] | origin_metadata, origin_metadata2 = sample_data.origin_metadata[:2] | ||||
| assert swh_storage.origin_add([origin]) == {"origin:add": 1} | assert swh_storage.origin_add([origin]) == {"origin:add": 1} | ||||
| swh_storage.metadata_fetcher_add([fetcher]) | swh_storage.metadata_fetcher_add([fetcher]) | ||||
| ▲ Show 20 Lines • Show All 547 Lines • Show Last 20 Lines | |||||