diff --git a/swh/graphql/tests/functional/test_branch_connection.py b/swh/graphql/tests/functional/test_branch_connection.py --- a/swh/graphql/tests/functional/test_branch_connection.py +++ b/swh/graphql/tests/functional/test_branch_connection.py @@ -5,16 +5,27 @@ 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 { @@ -37,14 +48,38 @@ } } } - """ - 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", [ @@ -56,40 +91,8 @@ ], ) 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 @@ -104,104 +107,32 @@ ], ) 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 diff --git a/swh/graphql/tests/functional/test_origin_connection.py b/swh/graphql/tests/functional/test_origin_connection.py --- a/swh/graphql/tests/functional/test_origin_connection.py +++ b/swh/graphql/tests/functional/test_origin_connection.py @@ -3,13 +3,13 @@ # 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) { @@ -25,7 +25,7 @@ """ % ( params, ) - return get_query_response(client, query_str) + return utils.get_query_response(client, query_str) def test_get(client, storage): diff --git a/swh/graphql/tests/functional/utils.py b/swh/graphql/tests/functional/utils.py --- a/swh/graphql/tests/functional/utils.py +++ b/swh/graphql/tests/functional/utils.py @@ -29,3 +29,8 @@ 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()])