Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/api/views/test_graph.py
| # Copyright (C) 2020 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 hashlib | import hashlib | ||||
| import json | import json | ||||
| import textwrap | import textwrap | ||||
| from hypothesis import given | |||||
| from django.http.response import StreamingHttpResponse | from django.http.response import StreamingHttpResponse | ||||
| from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import hash_to_bytes | ||||
| from swh.model.swhids import ExtendedObjectType, ExtendedSWHID | from swh.model.swhids import ExtendedObjectType, ExtendedSWHID | ||||
| from swh.web.api.views.graph import API_GRAPH_PERM | from swh.web.api.views.graph import API_GRAPH_PERM | ||||
| from swh.web.common.utils import reverse | from swh.web.common.utils import reverse | ||||
| from swh.web.config import SWH_WEB_INTERNAL_SERVER_NAME, get_config | from swh.web.config import SWH_WEB_INTERNAL_SERVER_NAME, get_config | ||||
| from swh.web.tests.strategies import origin | |||||
| from swh.web.tests.utils import check_http_get_response | from swh.web.tests.utils import check_http_get_response | ||||
| def test_graph_endpoint_no_authentication_for_vpn_users(api_client, requests_mock): | def test_graph_endpoint_no_authentication_for_vpn_users(api_client, requests_mock): | ||||
| graph_query = "stats" | graph_query = "stats" | ||||
| url = reverse("api-1-graph", url_args={"graph_query": graph_query}) | url = reverse("api-1-graph", url_args={"graph_query": graph_query}) | ||||
| requests_mock.get( | requests_mock.get( | ||||
| get_config()["graph"]["server_url"] + graph_query, | get_config()["graph"]["server_url"] + graph_query, | ||||
| ▲ Show 20 Lines • Show All 132 Lines • ▼ Show 20 Lines | def test_graph_ndjson_response(api_client, keycloak_oidc, requests_mock): | ||||
| url = reverse("api-1-graph", url_args={"graph_query": graph_query}) | url = reverse("api-1-graph", url_args={"graph_query": graph_query}) | ||||
| resp = check_http_get_response(api_client, url, status_code=200) | resp = check_http_get_response(api_client, url, status_code=200) | ||||
| assert isinstance(resp, StreamingHttpResponse) | assert isinstance(resp, StreamingHttpResponse) | ||||
| assert resp["Content-Type"] == "application/x-ndjson" | assert resp["Content-Type"] == "application/x-ndjson" | ||||
| assert b"".join(resp.streaming_content) == response_ndjson.encode() | assert b"".join(resp.streaming_content) == response_ndjson.encode() | ||||
| @given(origin()) | |||||
| def test_graph_response_resolve_origins( | def test_graph_response_resolve_origins( | ||||
| archive_data, api_client, keycloak_oidc, requests_mock, origin | archive_data, api_client, keycloak_oidc, requests_mock, origin | ||||
| ): | ): | ||||
| hasher = hashlib.sha1() | hasher = hashlib.sha1() | ||||
| hasher.update(origin["url"].encode()) | hasher.update(origin["url"].encode()) | ||||
| origin_sha1 = hasher.digest() | origin_sha1 = hasher.digest() | ||||
| origin_swhid = str( | origin_swhid = str( | ||||
| ExtendedSWHID(object_type=ExtendedObjectType.ORIGIN, object_id=origin_sha1) | ExtendedSWHID(object_type=ExtendedObjectType.ORIGIN, object_id=origin_sha1) | ||||
| ▲ Show 20 Lines • Show All 85 Lines • Show Last 20 Lines | |||||