Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/api/views/test_graph.py
Show All 10 Lines | |||||
from swh.model.identifiers import ORIGIN, SNAPSHOT, swhid | from swh.model.identifiers import ORIGIN, SNAPSHOT, swhid | ||||
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 get_config | from swh.web.config import get_config | ||||
from swh.web.tests.auth.keycloak_mock import mock_keycloak | from swh.web.tests.auth.keycloak_mock import mock_keycloak | ||||
from swh.web.tests.auth.sample_data import oidc_profile | from swh.web.tests.auth.sample_data import oidc_profile | ||||
from swh.web.tests.strategies import origin | from swh.web.tests.strategies import origin | ||||
from swh.web.tests.utils import check_http_get_response | |||||
def test_graph_endpoint_needs_authentication(api_client): | def test_graph_endpoint_needs_authentication(api_client): | ||||
url = reverse("api-1-graph", url_args={"graph_query": "stats"}) | url = reverse("api-1-graph", url_args={"graph_query": "stats"}) | ||||
resp = api_client.get(url) | check_http_get_response(api_client, url, status_code=401) | ||||
assert resp.status_code == 401 | |||||
def _authenticate_graph_user(api_client, mocker): | def _authenticate_graph_user(api_client, mocker): | ||||
mock_keycloak(mocker, user_permissions=[API_GRAPH_PERM]) | mock_keycloak(mocker, user_permissions=[API_GRAPH_PERM]) | ||||
api_client.credentials(HTTP_AUTHORIZATION=f"Bearer {oidc_profile['refresh_token']}") | api_client.credentials(HTTP_AUTHORIZATION=f"Bearer {oidc_profile['refresh_token']}") | ||||
def test_graph_endpoint_needs_permission(api_client, mocker, requests_mock): | def test_graph_endpoint_needs_permission(api_client, mocker, 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}) | ||||
api_client.credentials(HTTP_AUTHORIZATION=f"Bearer {oidc_profile['refresh_token']}") | api_client.credentials(HTTP_AUTHORIZATION=f"Bearer {oidc_profile['refresh_token']}") | ||||
mock_keycloak(mocker, user_permissions=[]) | mock_keycloak(mocker, user_permissions=[]) | ||||
resp = api_client.get(url) | check_http_get_response(api_client, url, status_code=403) | ||||
assert resp.status_code == 403 | |||||
_authenticate_graph_user(api_client, mocker) | _authenticate_graph_user(api_client, mocker) | ||||
requests_mock.get( | requests_mock.get( | ||||
get_config()["graph"]["server_url"] + graph_query, | get_config()["graph"]["server_url"] + graph_query, | ||||
json={}, | json={}, | ||||
headers={"Content-Type": "application/json"}, | headers={"Content-Type": "application/json"}, | ||||
) | ) | ||||
resp = api_client.get(url) | check_http_get_response(api_client, url, status_code=200) | ||||
assert resp.status_code == 200 | |||||
def test_graph_text_plain_response(api_client, mocker, requests_mock): | def test_graph_text_plain_response(api_client, mocker, requests_mock): | ||||
_authenticate_graph_user(api_client, mocker) | _authenticate_graph_user(api_client, mocker) | ||||
graph_query = "leaves/swh:1:dir:432d1b21c1256f7408a07c577b6974bbdbcc1323" | graph_query = "leaves/swh:1:dir:432d1b21c1256f7408a07c577b6974bbdbcc1323" | ||||
response_text = textwrap.dedent( | response_text = textwrap.dedent( | ||||
Show All 19 Lines | def test_graph_text_plain_response(api_client, mocker, requests_mock): | ||||
requests_mock.get( | requests_mock.get( | ||||
get_config()["graph"]["server_url"] + graph_query, | get_config()["graph"]["server_url"] + graph_query, | ||||
text=response_text, | text=response_text, | ||||
headers={"Content-Type": "text/plain"}, | headers={"Content-Type": "text/plain"}, | ||||
) | ) | ||||
url = reverse("api-1-graph", url_args={"graph_query": graph_query}) | url = reverse("api-1-graph", url_args={"graph_query": graph_query}) | ||||
resp = api_client.get(url) | resp = check_http_get_response( | ||||
api_client, url, status_code=200, content_type="text/plain" | |||||
assert resp.status_code == 200 | ) | ||||
assert resp.content_type == "text/plain" | |||||
assert resp.content == response_text.encode() | assert resp.content == response_text.encode() | ||||
_response_json = { | _response_json = { | ||||
"counts": {"nodes": 17075708289, "edges": 196236587976}, | "counts": {"nodes": 17075708289, "edges": 196236587976}, | ||||
"ratios": { | "ratios": { | ||||
"compression": 0.16, | "compression": 0.16, | ||||
"bits_per_node": 58.828, | "bits_per_node": 58.828, | ||||
Show All 13 Lines | def test_graph_json_response(api_client, mocker, requests_mock): | ||||
requests_mock.get( | requests_mock.get( | ||||
get_config()["graph"]["server_url"] + graph_query, | get_config()["graph"]["server_url"] + graph_query, | ||||
json=_response_json, | json=_response_json, | ||||
headers={"Content-Type": "application/json"}, | headers={"Content-Type": "application/json"}, | ||||
) | ) | ||||
url = reverse("api-1-graph", url_args={"graph_query": graph_query}) | url = reverse("api-1-graph", url_args={"graph_query": graph_query}) | ||||
resp = api_client.get(url) | resp = check_http_get_response(api_client, url, status_code=200) | ||||
assert resp.status_code == 200 | |||||
assert resp.content_type == "application/json" | assert resp.content_type == "application/json" | ||||
assert resp.content == json.dumps(_response_json).encode() | assert resp.content == json.dumps(_response_json).encode() | ||||
def test_graph_ndjson_response(api_client, mocker, requests_mock): | def test_graph_ndjson_response(api_client, mocker, requests_mock): | ||||
_authenticate_graph_user(api_client, mocker) | _authenticate_graph_user(api_client, mocker) | ||||
graph_query = "visit/paths/swh:1:dir:644dd466d8ad527ea3a609bfd588a3244e6dafcb" | graph_query = "visit/paths/swh:1:dir:644dd466d8ad527ea3a609bfd588a3244e6dafcb" | ||||
Show All 12 Lines | def test_graph_ndjson_response(api_client, mocker, requests_mock): | ||||
requests_mock.get( | requests_mock.get( | ||||
get_config()["graph"]["server_url"] + graph_query, | get_config()["graph"]["server_url"] + graph_query, | ||||
text=response_ndjson, | text=response_ndjson, | ||||
headers={"Content-Type": "application/x-ndjson"}, | headers={"Content-Type": "application/x-ndjson"}, | ||||
) | ) | ||||
url = reverse("api-1-graph", url_args={"graph_query": graph_query}) | url = reverse("api-1-graph", url_args={"graph_query": graph_query}) | ||||
resp = api_client.get(url) | resp = check_http_get_response(api_client, url, status_code=200) | ||||
assert resp.status_code == 200 | |||||
assert resp.content_type == "application/x-ndjson" | assert resp.content_type == "application/x-ndjson" | ||||
assert resp.content == response_ndjson.encode() | assert resp.content == response_ndjson.encode() | ||||
@given(origin()) | @given(origin()) | ||||
def test_graph_response_resolve_origins( | def test_graph_response_resolve_origins( | ||||
archive_data, api_client, mocker, requests_mock, origin | archive_data, api_client, mocker, requests_mock, origin | ||||
): | ): | ||||
Show All 34 Lines | ): | ||||
) | ) | ||||
url = reverse( | url = reverse( | ||||
"api-1-graph", | "api-1-graph", | ||||
url_args={"graph_query": graph_query}, | url_args={"graph_query": graph_query}, | ||||
query_params={"direction": "backward"}, | query_params={"direction": "backward"}, | ||||
) | ) | ||||
resp = api_client.get(url) | resp = check_http_get_response(api_client, url, status_code=200) | ||||
assert resp.status_code == 200 | |||||
assert resp.content_type == content_type | assert resp.content_type == content_type | ||||
assert resp.content == response_text.encode() | assert resp.content == response_text.encode() | ||||
url = reverse( | url = reverse( | ||||
"api-1-graph", | "api-1-graph", | ||||
url_args={"graph_query": graph_query}, | url_args={"graph_query": graph_query}, | ||||
query_params={"direction": "backward", "resolve_origins": "true"}, | query_params={"direction": "backward", "resolve_origins": "true"}, | ||||
) | ) | ||||
resp = api_client.get(url) | resp = check_http_get_response(api_client, url, status_code=200) | ||||
assert resp.status_code == 200 | |||||
assert resp.content_type == content_type | assert resp.content_type == content_type | ||||
assert ( | assert ( | ||||
resp.content == response_text.replace(origin_swhid, origin["url"]).encode() | resp.content == response_text.replace(origin_swhid, origin["url"]).encode() | ||||
) | ) | ||||
def test_graph_response_resolve_origins_nothing_to_do( | def test_graph_response_resolve_origins_nothing_to_do( | ||||
api_client, mocker, requests_mock | api_client, mocker, requests_mock | ||||
Show All 9 Lines | ): | ||||
) | ) | ||||
url = reverse( | url = reverse( | ||||
"api-1-graph", | "api-1-graph", | ||||
url_args={"graph_query": graph_query}, | url_args={"graph_query": graph_query}, | ||||
query_params={"resolve_origins": "true"}, | query_params={"resolve_origins": "true"}, | ||||
) | ) | ||||
resp = api_client.get(url) | resp = check_http_get_response(api_client, url, status_code=200) | ||||
assert resp.status_code == 200 | |||||
assert resp.content_type == "application/json" | assert resp.content_type == "application/json" | ||||
assert resp.content == json.dumps(_response_json).encode() | assert resp.content == json.dumps(_response_json).encode() |