Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/api/views/test_metadata.py
| # Copyright (C) 2021 The Software Heritage developers | # Copyright (C) 2021 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU Affero General Public License version 3, or any later version | # License: GNU Affero General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| import attr | import attr | ||||
| from hypothesis import given, strategies | from hypothesis import given, strategies | ||||
| import pytest | import pytest | ||||
| from swh.model.hypothesis_strategies import raw_extrinsic_metadata | from swh.model.hypothesis_strategies import raw_extrinsic_metadata | ||||
| from swh.web.common.utils import reverse | from swh.web.common.utils import reverse | ||||
| from swh.web.tests.api.views.utils import scroll_results | from swh.web.tests.api.views.utils import scroll_results | ||||
| from swh.web.tests.utils import check_api_get_responses, check_http_get_response | from swh.web.tests.utils import check_api_get_responses, check_http_get_response | ||||
| @given(raw_extrinsic_metadata()) | @given(raw_extrinsic_metadata()) | ||||
| def test_api_raw_extrinsic_metadata(api_client, archive_data, metadata): | def test_api_raw_extrinsic_metadata(api_client, subtest, metadata): | ||||
| # ensure archive_data fixture will be reset between each hypothesis | |||||
| # example test run | |||||
| @subtest | |||||
| def test_inner(archive_data): | |||||
| archive_data.metadata_authority_add([metadata.authority]) | archive_data.metadata_authority_add([metadata.authority]) | ||||
| archive_data.metadata_fetcher_add([metadata.fetcher]) | archive_data.metadata_fetcher_add([metadata.fetcher]) | ||||
| archive_data.raw_extrinsic_metadata_add([metadata]) | archive_data.raw_extrinsic_metadata_add([metadata]) | ||||
| authority = metadata.authority | authority = metadata.authority | ||||
| url = reverse( | url = reverse( | ||||
| "api-1-raw-extrinsic-metadata-swhid", | "api-1-raw-extrinsic-metadata-swhid", | ||||
| url_args={"target": str(metadata.target)}, | url_args={"target": str(metadata.target)}, | ||||
| query_params={"authority": f"{authority.type.value} {authority.url}"}, | query_params={"authority": f"{authority.type.value} {authority.url}"}, | ||||
| ) | ) | ||||
| rv = check_api_get_responses(api_client, url, status_code=200) | rv = check_api_get_responses(api_client, url, status_code=200) | ||||
| assert len(rv.data) == 1 | assert len(rv.data) == 1 | ||||
| expected_result = metadata.to_dict() | expected_result = metadata.to_dict() | ||||
| del expected_result["id"] | del expected_result["id"] | ||||
| del expected_result["metadata"] | del expected_result["metadata"] | ||||
| metadata_url = rv.data[0]["metadata_url"] | metadata_url = rv.data[0]["metadata_url"] | ||||
| expected_result["metadata_url"] = metadata_url | expected_result["metadata_url"] = metadata_url | ||||
| expected_result["discovery_date"] = expected_result["discovery_date"].isoformat() | expected_result["discovery_date"] = expected_result[ | ||||
| "discovery_date" | |||||
| ].isoformat() | |||||
| assert rv.data == [expected_result] | assert rv.data == [expected_result] | ||||
| rv = check_http_get_response(api_client, metadata_url, status_code=200) | rv = check_http_get_response(api_client, metadata_url, status_code=200) | ||||
| assert rv["Content-Type"] == "application/octet-stream" | assert rv["Content-Type"] == "application/octet-stream" | ||||
| assert ( | assert ( | ||||
| rv["Content-Disposition"] | rv["Content-Disposition"] | ||||
| == f'attachment; filename="{metadata.target}_metadata"' | == f'attachment; filename="{metadata.target}_metadata"' | ||||
| ) | ) | ||||
| assert rv.content == metadata.metadata | assert rv.content == metadata.metadata | ||||
| @pytest.mark.parametrize("limit", [1, 2, 10, 100]) | @pytest.mark.parametrize("limit", [1, 2, 10, 100]) | ||||
| @given(strategies.sets(raw_extrinsic_metadata(), min_size=1)) | @given(strategies.sets(raw_extrinsic_metadata(), min_size=1)) | ||||
| def test_api_raw_extrinsic_metadata_scroll(api_client, archive_data, limit, metadata): | def test_api_raw_extrinsic_metadata_scroll(api_client, subtest, limit, meta): | ||||
| # ensure archive_data fixture will be reset between each hypothesis | |||||
| # example test run | |||||
| @subtest | |||||
| def test_inner(archive_data): | |||||
| # Make all metadata objects use the same authority and target | # Make all metadata objects use the same authority and target | ||||
| metadata0 = next(iter(metadata)) | metadata0 = next(iter(meta)) | ||||
| metadata = { | metadata = { | ||||
| attr.evolve(m, authority=metadata0.authority, target=metadata0.target) | attr.evolve(m, authority=metadata0.authority, target=metadata0.target) | ||||
| for m in metadata | for m in meta | ||||
| } | } | ||||
| # Metadata ids must also be updated as they depend on authority and target | |||||
| metadata = {attr.evolve(m, id=m.compute_hash()) for m in metadata} | |||||
| authority = metadata0.authority | authority = metadata0.authority | ||||
| archive_data.metadata_authority_add([authority]) | archive_data.metadata_authority_add([authority]) | ||||
| archive_data.metadata_fetcher_add(list({m.fetcher for m in metadata})) | archive_data.metadata_fetcher_add(list({m.fetcher for m in metadata})) | ||||
| archive_data.raw_extrinsic_metadata_add(metadata) | archive_data.raw_extrinsic_metadata_add(metadata) | ||||
| url = reverse( | url = reverse( | ||||
| "api-1-raw-extrinsic-metadata-swhid", | "api-1-raw-extrinsic-metadata-swhid", | ||||
| url_args={"target": str(metadata0.target)}, | url_args={"target": str(metadata0.target)}, | ||||
| query_params={ | query_params={ | ||||
| "authority": f"{authority.type.value} {authority.url}", | "authority": f"{authority.type.value} {authority.url}", | ||||
| "limit": limit, | "limit": limit, | ||||
| }, | }, | ||||
| ) | ) | ||||
| results = scroll_results(api_client, url) | results = scroll_results(api_client, url) | ||||
| expected_results = [m.to_dict() for m in metadata] | expected_results = [m.to_dict() for m in metadata] | ||||
| for expected_result in expected_results: | for expected_result in expected_results: | ||||
| del expected_result["id"] | del expected_result["id"] | ||||
| del expected_result["metadata"] | del expected_result["metadata"] | ||||
| expected_result["discovery_date"] = expected_result[ | expected_result["discovery_date"] = expected_result[ | ||||
| "discovery_date" | "discovery_date" | ||||
| ].isoformat() | ].isoformat() | ||||
| assert len(results) == len(expected_results) | |||||
| for result in results: | for result in results: | ||||
| del result["metadata_url"] | del result["metadata_url"] | ||||
| assert result in expected_results | |||||
| assert results == expected_results | |||||
| _swhid = "swh:1:dir:a2faa28028657859c16ff506924212b33f0e1307" | _swhid = "swh:1:dir:a2faa28028657859c16ff506924212b33f0e1307" | ||||
| @pytest.mark.parametrize( | @pytest.mark.parametrize( | ||||
| "status_code,url_args,query_params", | "status_code,url_args,query_params", | ||||
| [ | [ | ||||
| Show All 40 Lines | url = reverse( | ||||
| "api-1-raw-extrinsic-metadata-swhid", | "api-1-raw-extrinsic-metadata-swhid", | ||||
| url_args=url_args, | url_args=url_args, | ||||
| query_params=query_params, | query_params=query_params, | ||||
| ) | ) | ||||
| check_api_get_responses(api_client, url, status_code=status_code) | check_api_get_responses(api_client, url, status_code=status_code) | ||||
| @given(raw_extrinsic_metadata()) | @given(raw_extrinsic_metadata()) | ||||
| def test_api_raw_extrinsic_metadata_list_authorities( | def test_api_raw_extrinsic_metadata_list_authorities(api_client, subtest, metadata): | ||||
| api_client, archive_data, metadata | # ensure archive_data fixture will be reset between each hypothesis | ||||
| ): | # example test run | ||||
| @subtest | |||||
| def test_inner(archive_data): | |||||
| archive_data.metadata_authority_add([metadata.authority]) | archive_data.metadata_authority_add([metadata.authority]) | ||||
| archive_data.metadata_fetcher_add([metadata.fetcher]) | archive_data.metadata_fetcher_add([metadata.fetcher]) | ||||
| archive_data.raw_extrinsic_metadata_add([metadata]) | archive_data.raw_extrinsic_metadata_add([metadata]) | ||||
| authority = metadata.authority | authority = metadata.authority | ||||
| url = reverse( | url = reverse( | ||||
| "api-1-raw-extrinsic-metadata-swhid-authorities", | "api-1-raw-extrinsic-metadata-swhid-authorities", | ||||
| url_args={"target": str(metadata.target)}, | url_args={"target": str(metadata.target)}, | ||||
| ) | ) | ||||
| rv = check_api_get_responses(api_client, url, status_code=200) | rv = check_api_get_responses(api_client, url, status_code=200) | ||||
| expected_results = [ | expected_results = [ | ||||
| { | { | ||||
| "type": authority.type.value, | "type": authority.type.value, | ||||
| "url": authority.url, | "url": authority.url, | ||||
| "metadata_list_url": "http://testserver" | "metadata_list_url": "http://testserver" | ||||
| + reverse( | + reverse( | ||||
| "api-1-raw-extrinsic-metadata-swhid", | "api-1-raw-extrinsic-metadata-swhid", | ||||
| url_args={"target": str(metadata.target)}, | url_args={"target": str(metadata.target)}, | ||||
| query_params={"authority": f"{authority.type.value} {authority.url}"}, | query_params={ | ||||
| "authority": f"{authority.type.value} {authority.url}" | |||||
| }, | |||||
| ), | ), | ||||
| } | } | ||||
| ] | ] | ||||
| assert rv.data == expected_results | assert rv.data == expected_results | ||||