Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/api/views/test_graph.py
Show All 32 Lines | def test_graph_endpoint_no_authentication_for_vpn_users(api_client, requests_mock): | ||||
) | ) | ||||
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"}) | ||||
check_http_get_response(api_client, url, status_code=401) | check_http_get_response(api_client, url, status_code=401) | ||||
def _authenticate_graph_user(api_client, keycloak_mock): | def _authenticate_graph_user(api_client, keycloak_oidc): | ||||
keycloak_mock.user_permissions = [API_GRAPH_PERM] | keycloak_oidc.user_permissions = [API_GRAPH_PERM] | ||||
oidc_profile = keycloak_mock.login() | oidc_profile = keycloak_oidc.login() | ||||
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, keycloak_mock, requests_mock): | def test_graph_endpoint_needs_permission(api_client, keycloak_oidc, 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}) | ||||
oidc_profile = keycloak_mock.login() | oidc_profile = keycloak_oidc.login() | ||||
api_client.credentials(HTTP_AUTHORIZATION=f"Bearer {oidc_profile['refresh_token']}") | api_client.credentials(HTTP_AUTHORIZATION=f"Bearer {oidc_profile['refresh_token']}") | ||||
check_http_get_response(api_client, url, status_code=403) | check_http_get_response(api_client, url, status_code=403) | ||||
_authenticate_graph_user(api_client, keycloak_mock) | _authenticate_graph_user(api_client, keycloak_oidc) | ||||
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"}, | ||||
) | ) | ||||
check_http_get_response(api_client, url, status_code=200) | check_http_get_response(api_client, url, status_code=200) | ||||
def test_graph_text_plain_response(api_client, keycloak_mock, requests_mock): | def test_graph_text_plain_response(api_client, keycloak_oidc, requests_mock): | ||||
_authenticate_graph_user(api_client, keycloak_mock) | _authenticate_graph_user(api_client, keycloak_oidc) | ||||
graph_query = "leaves/swh:1:dir:432d1b21c1256f7408a07c577b6974bbdbcc1323" | graph_query = "leaves/swh:1:dir:432d1b21c1256f7408a07c577b6974bbdbcc1323" | ||||
response_text = textwrap.dedent( | response_text = textwrap.dedent( | ||||
"""\ | """\ | ||||
swh:1:cnt:1d3dace0a825b0535c37c53ed669ef817e9c1b47 | swh:1:cnt:1d3dace0a825b0535c37c53ed669ef817e9c1b47 | ||||
swh:1:cnt:6d5b280f4e33589ae967a7912a587dd5cb8dedaa | swh:1:cnt:6d5b280f4e33589ae967a7912a587dd5cb8dedaa | ||||
swh:1:cnt:91bef238bf01356a550d416d14bb464c576ac6f4 | swh:1:cnt:91bef238bf01356a550d416d14bb464c576ac6f4 | ||||
Show All 35 Lines | "ratios": { | ||||
"bits_per_edge": 5.119, | "bits_per_edge": 5.119, | ||||
"avg_locality": 2184278529.729, | "avg_locality": 2184278529.729, | ||||
}, | }, | ||||
"indegree": {"min": 0, "max": 263180117, "avg": 11.4921492364925}, | "indegree": {"min": 0, "max": 263180117, "avg": 11.4921492364925}, | ||||
"outdegree": {"min": 0, "max": 1033207, "avg": 11.4921492364925}, | "outdegree": {"min": 0, "max": 1033207, "avg": 11.4921492364925}, | ||||
} | } | ||||
def test_graph_json_response(api_client, keycloak_mock, requests_mock): | def test_graph_json_response(api_client, keycloak_oidc, requests_mock): | ||||
_authenticate_graph_user(api_client, keycloak_mock) | _authenticate_graph_user(api_client, keycloak_oidc) | ||||
graph_query = "stats" | graph_query = "stats" | ||||
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 = check_http_get_response(api_client, url, status_code=200) | resp = check_http_get_response(api_client, url, 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, keycloak_mock, requests_mock): | def test_graph_ndjson_response(api_client, keycloak_oidc, requests_mock): | ||||
_authenticate_graph_user(api_client, keycloak_mock) | _authenticate_graph_user(api_client, keycloak_oidc) | ||||
graph_query = "visit/paths/swh:1:dir:644dd466d8ad527ea3a609bfd588a3244e6dafcb" | graph_query = "visit/paths/swh:1:dir:644dd466d8ad527ea3a609bfd588a3244e6dafcb" | ||||
response_ndjson = textwrap.dedent( | response_ndjson = textwrap.dedent( | ||||
"""\ | """\ | ||||
["swh:1:dir:644dd466d8ad527ea3a609bfd588a3244e6dafcb",\ | ["swh:1:dir:644dd466d8ad527ea3a609bfd588a3244e6dafcb",\ | ||||
"swh:1:cnt:acfb7cabd63b368a03a9df87670ece1488c8bce0"] | "swh:1:cnt:acfb7cabd63b368a03a9df87670ece1488c8bce0"] | ||||
["swh:1:dir:644dd466d8ad527ea3a609bfd588a3244e6dafcb",\ | ["swh:1:dir:644dd466d8ad527ea3a609bfd588a3244e6dafcb",\ | ||||
Show All 17 Lines | def test_graph_ndjson_response(api_client, keycloak_oidc, requests_mock): | ||||
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()) | @given(origin()) | ||||
def test_graph_response_resolve_origins( | def test_graph_response_resolve_origins( | ||||
archive_data, api_client, keycloak_mock, 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) | ||||
) | ) | ||||
snapshot = archive_data.snapshot_get_latest(origin["url"])["id"] | snapshot = archive_data.snapshot_get_latest(origin["url"])["id"] | ||||
snapshot_swhid = str( | snapshot_swhid = str( | ||||
ExtendedSWHID( | ExtendedSWHID( | ||||
object_type=ExtendedObjectType.SNAPSHOT, object_id=hash_to_bytes(snapshot) | object_type=ExtendedObjectType.SNAPSHOT, object_id=hash_to_bytes(snapshot) | ||||
) | ) | ||||
) | ) | ||||
_authenticate_graph_user(api_client, keycloak_mock) | _authenticate_graph_user(api_client, keycloak_oidc) | ||||
for graph_query, response_text, content_type in ( | for graph_query, response_text, content_type in ( | ||||
( | ( | ||||
f"visit/nodes/{snapshot_swhid}", | f"visit/nodes/{snapshot_swhid}", | ||||
f"{snapshot_swhid}\n{origin_swhid}\n", | f"{snapshot_swhid}\n{origin_swhid}\n", | ||||
"text/plain", | "text/plain", | ||||
), | ), | ||||
( | ( | ||||
Show All 39 Lines | ): | ||||
assert resp["Content-Type"] == content_type | assert resp["Content-Type"] == content_type | ||||
assert ( | assert ( | ||||
b"".join(resp.streaming_content) | b"".join(resp.streaming_content) | ||||
== response_text.replace(origin_swhid, origin["url"]).encode() | == 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, keycloak_mock, requests_mock | api_client, keycloak_oidc, requests_mock | ||||
): | ): | ||||
_authenticate_graph_user(api_client, keycloak_mock) | _authenticate_graph_user(api_client, keycloak_oidc) | ||||
graph_query = "stats" | graph_query = "stats" | ||||
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"}, | ||||
) | ) | ||||
Show All 10 Lines |