diff --git a/swh/graphql/tests/functional/test_branch_connection.py b/swh/graphql/tests/functional/test_branch_connection.py index 41fe5aa..0cb04f9 100644 --- a/swh/graphql/tests/functional/test_branch_connection.py +++ b/swh/graphql/tests/functional/test_branch_connection.py @@ -1,209 +1,140 @@ # Copyright (C) 2022 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information import pytest -from .utils import get_query_response +from . import utils -def test_get(client): +def get_branches(client, swhid: str, first: int, **args) -> tuple: + args["first"] = first + params = utils.get_query_params_from_args(**args) query_str = """ { - snapshot(swhid: "swh:1:snp:0e7f84ede9a254f2cd55649ad5240783f557e65f") { - branches(first:10) { + snapshot(swhid: "%s") { + branches(%s) { + pageInfo { + endCursor + } + edges { + cursor + } nodes { type + name { + text + } target { __typename ...on Revision { swhid } ...on Release { swhid } ...on Content { swhid } ...on Directory { swhid } ...on Snapshot { swhid } } } } } } - """ - data, errors = get_query_response(client, query_str) + """ % ( + swhid, + params, + ) + return utils.get_query_response(client, query_str) + + +def test_get(client): + swhid = "swh:1:snp:0e7f84ede9a254f2cd55649ad5240783f557e65f" + data, errors = get_branches(client, swhid, 10) # Alias type is not handled at the moment, hence the error assert len(errors) == 1 assert errors[0]["message"] == "Invalid node type: branch-alias" assert len(data["snapshot"]["branches"]["nodes"]) == 5 +def test_get_data(client): + swhid = "swh:1:snp:0e7f84ede9a254f2cd55649ad5240783f557e65f" + data, errors = get_branches(client, swhid, 10, types="[revision]") + assert len(data["snapshot"]["branches"]["nodes"]) == 1 + # filter 'type' will return a single revision object and is used to assert data + node = data["snapshot"]["branches"]["nodes"][0] + assert node == { + "name": {"text": "target/revision"}, + "target": { + "__typename": "Revision", + "swhid": "swh:1:rev:66c7c1cd9673275037140f2abff7b7b11fc9439c", + }, + "type": "revision", + } + + @pytest.mark.parametrize( "filter_type, count, target_type, swhid_pattern", [ ("revision", 1, "Revision", "swh:1:rev"), ("release", 1, "Release", "swh:1:rel"), ("directory", 1, "Directory", "swh:1:dir"), ("content", 0, "Content", "swh:1:cnt"), ("snapshot", 1, "Snapshot", "swh:1:snp"), ], ) def test_get_type_filter(client, filter_type, count, target_type, swhid_pattern): - query_str = ( - """ - { - snapshot(swhid: "swh:1:snp:0e7f84ede9a254f2cd55649ad5240783f557e65f") { - branches(first:10, types: [%s]) { - nodes { - type - target { - __typename - ...on Revision { - swhid - } - ...on Release { - swhid - } - ...on Content { - swhid - } - ...on Directory { - swhid - } - ...on Snapshot { - swhid - } - } - } - } - } - } - """ - % filter_type - ) - data, _ = get_query_response(client, query_str) - + swhid = "swh:1:snp:0e7f84ede9a254f2cd55649ad5240783f557e65f" + data, _ = get_branches(client, swhid, 10, types=f"[{filter_type}]") assert len(data["snapshot"]["branches"]["nodes"]) == count for node in data["snapshot"]["branches"]["nodes"]: assert node["target"]["__typename"] == target_type assert node["target"]["swhid"].startswith(swhid_pattern) @pytest.mark.parametrize( "filter_types, count", [ ("revision, release", 2), ("revision, snapshot, release", 3), ], ) def test_get_type_filter_multiple(client, filter_types, count): - query_str = ( - """ - { - snapshot(swhid: "swh:1:snp:0e7f84ede9a254f2cd55649ad5240783f557e65f") { - branches(first:10, types: [%s]) { - nodes { - type - } - } - } - }""" - % filter_types - ) - data, _ = get_query_response(client, query_str) + swhid = "swh:1:snp:0e7f84ede9a254f2cd55649ad5240783f557e65f" + data, _ = get_branches(client, swhid, 10, types=f"[{filter_types}]") assert len(data["snapshot"]["branches"]["nodes"]) == count @pytest.mark.parametrize("name", ["rel", "rev", "non-exist"]) def test_get_name_include_filter(client, name): - query_str = ( - """ - { - snapshot(swhid: "swh:1:snp:0e7f84ede9a254f2cd55649ad5240783f557e65f") { - branches(first:10, nameInclude: "%s") { - nodes { - name { - text - } - } - } - } - }""" - % name - ) - data, _ = get_query_response(client, query_str) + swhid = "swh:1:snp:0e7f84ede9a254f2cd55649ad5240783f557e65f" + data, _ = get_branches(client, swhid, 10, nameInclude=f'"{name}"') for node in data["snapshot"]["branches"]["nodes"]: assert name in node["name"]["text"] @pytest.mark.parametrize("count", [1, 2]) def test_get_first_arg(client, count): - query_str = ( - """ - { - snapshot(swhid: "swh:1:snp:0e7f84ede9a254f2cd55649ad5240783f557e65f") { - branches(first: %s) { - nodes { - type - } - } - } - }""" - % count - ) - data, _ = get_query_response(client, query_str) + swhid = "swh:1:snp:0e7f84ede9a254f2cd55649ad5240783f557e65f" + data, _ = get_branches(client, swhid, first=count) assert len(data["snapshot"]["branches"]["nodes"]) == count def test_get_after_arg(client): - query_str = """ - { - snapshot(swhid: "swh:1:snp:0e7f84ede9a254f2cd55649ad5240783f557e65f") { - branches(first: 1) { - pageInfo { - endCursor - } - nodes { - name { - text - } - } - } - } - }""" - first_data, _ = get_query_response(client, query_str) + swhid = "swh:1:snp:0e7f84ede9a254f2cd55649ad5240783f557e65f" + first_data, _ = get_branches(client, swhid, first=1) end_cursor = first_data["snapshot"]["branches"]["pageInfo"]["endCursor"] node_name = first_data["snapshot"]["branches"]["nodes"][0]["name"]["text"] - - query_str = ( - """ - { - snapshot(swhid: "swh:1:snp:0e7f84ede9a254f2cd55649ad5240783f557e65f") { - branches(first: 3, after: "%s") { - nodes { - type - name { - text - } - } - edges { - cursor - } - } - } - }""" - % end_cursor - ) - second_data, _ = get_query_response(client, query_str) + second_data, _ = get_branches(client, swhid, first=3, after=f'"{end_cursor}"') branches = second_data["snapshot"]["branches"] assert len(branches["nodes"]) == 3 assert branches["edges"][0]["cursor"] == end_cursor for node in branches["nodes"]: assert node["name"]["text"] > node_name diff --git a/swh/graphql/tests/functional/test_origin_connection.py b/swh/graphql/tests/functional/test_origin_connection.py index d128958..1af79fc 100644 --- a/swh/graphql/tests/functional/test_origin_connection.py +++ b/swh/graphql/tests/functional/test_origin_connection.py @@ -1,49 +1,49 @@ # Copyright (C) 2022 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information +from . import utils from ..data import get_origins -from .utils import get_query_response def get_origins_from_api(client, first: int, **args) -> tuple: args["first"] = first - params = ",".join([f"{key}: {val}" for (key, val) in args.items()]) + params = utils.get_query_params_from_args(**args) query_str = """ { origins(%s) { nodes { url } pageInfo { hasNextPage endCursor } } } """ % ( params, ) - return get_query_response(client, query_str) + return utils.get_query_response(client, query_str) def test_get(client, storage): data, _ = get_origins_from_api(client, 10) assert len(data["origins"]["nodes"]) == len(get_origins()) def test_get_filter_by_pattern(client): data, _ = get_origins_from_api(client, 10, urlPattern='"somewhere.org/den"') assert len(data["origins"]["nodes"]) == 1 def test_get_filter_by_non_existing_pattern(client): data, _ = get_origins_from_api(client, 10, urlPattern='"somewhere.org/den/test/"') assert len(data["origins"]["nodes"]) == 0 def test_basic_pagination(client): data, _ = get_origins_from_api(client, first=len(get_origins())) assert len(data["origins"]["nodes"]) == len(get_origins()) assert data["origins"]["pageInfo"] == {"hasNextPage": False, "endCursor": None} diff --git a/swh/graphql/tests/functional/utils.py b/swh/graphql/tests/functional/utils.py index 80490dd..43207dd 100644 --- a/swh/graphql/tests/functional/utils.py +++ b/swh/graphql/tests/functional/utils.py @@ -1,31 +1,36 @@ # Copyright (C) 2022 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information import json from typing import Dict, Tuple def get_response(client, query_str: str): return client.post("/", json={"query": query_str}) def get_query_response(client, query_str: str) -> Tuple[Dict, Dict]: response = get_response(client, query_str) assert response.status_code == 200, response.data result = json.loads(response.data) return result.get("data"), result.get("errors") def assert_missing_object(client, query_str: str, obj_type: str) -> None: data, errors = get_query_response(client, query_str) assert data[obj_type] is None assert len(errors) == 1 assert errors[0]["message"] == "Requested object is not available" def get_error_response(client, query_str: str, error_code: int = 400) -> Dict: response = get_response(client, query_str) assert response.status_code == error_code return json.loads(response.data)["errors"] + + +def get_query_params_from_args(**args) -> str: + # build a GraphQL query parameters string from arguments + return ",".join([f"{key}: {val}" for (key, val) in args.items()])