diff --git a/swh/graphql/resolvers/directory.py b/swh/graphql/resolvers/directory.py index 7c3e126..987575d 100644 --- a/swh/graphql/resolvers/directory.py +++ b/swh/graphql/resolvers/directory.py @@ -1,76 +1,75 @@ # 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 typing import Union from swh.graphql.backends import archive from swh.model.model import Directory from swh.model.swhids import ObjectType from .base_node import BaseSWHNode from .release import BaseReleaseNode from .revision import BaseRevisionNode from .snapshot_branch import SnapshotBranchNode class BaseDirectoryNode(BaseSWHNode): """ Base resolver for all the directory nodes """ def _get_directory_by_id(self, directory_id): # Return a Directory model object # entries is initialized as empty # Same pattern is used in snapshot return Directory(id=directory_id, entries=()) def is_type_of(self): return "Directory" class DirectoryNode(BaseDirectoryNode): """ Node resolver for a directory requested directly with its SWHID """ def _get_node_data(self): swhid = self.kwargs.get("swhid") # path = "" if ( swhid.object_type == ObjectType.DIRECTORY and archive.Archive().is_object_available( swhid.object_id, swhid.object_type ) ): # _get_directory_by_id is not making any backend call # hence the is_directory_available validation return self._get_directory_by_id(swhid.object_id) return None class RevisionDirectoryNode(BaseDirectoryNode): """ Node resolver for a directory requested from a revision """ obj: BaseRevisionNode def _get_node_data(self): - # self.obj.directory_swhid is the requested directory SWHID - directory_id = self.obj.directory_swhid.object_id - return self._get_directory_by_id(directory_id) + # self.obj.directory_hash is the requested directory Id + return self._get_directory_by_id(self.obj.directory_hash) class TargetDirectoryNode(BaseDirectoryNode): """ Node resolver for a directory requested as a target """ from .directory_entry import DirectoryEntryNode obj: Union[SnapshotBranchNode, BaseReleaseNode, DirectoryEntryNode] def _get_node_data(self): return self._get_directory_by_id(self.obj.target_hash) diff --git a/swh/graphql/resolvers/revision.py b/swh/graphql/resolvers/revision.py index 7272607..d2058d5 100644 --- a/swh/graphql/resolvers/revision.py +++ b/swh/graphql/resolvers/revision.py @@ -1,108 +1,106 @@ # 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 typing import Union from swh.graphql.backends import archive from swh.graphql.utils import utils from swh.model.swhids import CoreSWHID, ObjectType from swh.storage.interface import PagedResult from .base_connection import BaseConnection from .base_node import BaseSWHNode from .release import BaseReleaseNode from .snapshot_branch import SnapshotBranchNode class BaseRevisionNode(BaseSWHNode): """ Base resolver for all the revision nodes """ def _get_revision_by_id(self, revision_id): return (archive.Archive().get_revisions([revision_id]) or None)[0] @property def parent_swhids(self): # for ParentRevisionConnection resolver return [ CoreSWHID(object_type=ObjectType.REVISION, object_id=parent_id) for parent_id in self._node.parents ] @property - def directory_swhid(self): # for RevisionDirectoryNode resolver - return CoreSWHID( - object_type=ObjectType.DIRECTORY, object_id=self._node.directory - ) + def directory_hash(self): # for RevisionDirectoryNode resolver + return self._node.directory @property def type(self): return self._node.type.value def is_type_of(self): # is_type_of is required only when resolving a UNION type # This is for ariadne to return the right type return "Revision" class RevisionNode(BaseRevisionNode): """ Node resolver for a revision requested directly with its SWHID """ def _get_node_data(self): return self._get_revision_by_id(self.kwargs.get("swhid").object_id) class TargetRevisionNode(BaseRevisionNode): """ Node resolver for a revision requested as a target """ obj: Union[SnapshotBranchNode, BaseReleaseNode] def _get_node_data(self): # self.obj.target_hash is the requested revision id return self._get_revision_by_id(self.obj.target_hash) class ParentRevisionConnection(BaseConnection): """ Connection resolver for parent revisions in a revision """ obj: BaseRevisionNode _node_class = BaseRevisionNode def _get_paged_result(self) -> PagedResult: # self.obj is the current(child) revision # self.obj.parent_swhids is the list of parent SWHIDs # FIXME, using dummy(local) pagination, move pagination to backend # To remove localpagination, just drop the paginated call # STORAGE-TODO (pagination) parents = archive.Archive().get_revisions( [x.object_id for x in self.obj.parent_swhids] ) return utils.paginated(parents, self._get_first_arg(), self._get_after_arg()) class LogRevisionConnection(BaseConnection): """ Connection resolver for the log (list of revisions) in a revision """ obj: BaseRevisionNode _node_class = BaseRevisionNode def _get_paged_result(self) -> PagedResult: # STORAGE-TODO (date in revisionlog is a dict) log = archive.Archive().get_revision_log([self.obj.swhid.object_id]) # FIXME, using dummy(local) pagination, move pagination to backend # To remove localpagination, just drop the paginated call # STORAGE-TODO (pagination) return utils.paginated(log, self._get_first_arg(), self._get_after_arg()) diff --git a/swh/graphql/tests/functional/test_directory.py b/swh/graphql/tests/functional/test_directory.py new file mode 100644 index 0000000..494d73b --- /dev/null +++ b/swh/graphql/tests/functional/test_directory.py @@ -0,0 +1,74 @@ +# 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 . import utils +from ..data import get_directories + + +@pytest.mark.parametrize("directory", get_directories()) +def test_get_directory(client, directory): + query_str = """ + { + directory(swhid: "%s") { + swhid + } + } + """ + data, _ = utils.get_query_response(client, query_str % directory.swhid()) + assert data["directory"] == {"swhid": str(directory.swhid())} + + +def test_get_directory_with_invalid_swhid(client): + query_str = """ + { + directory(swhid: "swh:1:dir:invalid") { + swhid + } + } + """ + errors = utils.get_error_response(client, query_str) + # API will throw an error in case of an invalid SWHID + assert len(errors) == 1 + assert "Invalid SWHID: invalid syntax" in errors[0]["message"] + + +def test_get_revision_directory(client): + query_str = """ + { + revision(swhid: "swh:1:rev:66c7c1cd9673275037140f2abff7b7b11fc9439c") { + swhid + directory { + swhid + } + } + } + """ + data, _ = utils.get_query_response(client, query_str) + assert data["revision"]["directory"] == { + "swhid": "swh:1:dir:0101010101010101010101010101010101010101" + } + + +def test_get_target_directory(client): + # TargetDirectoryNode is returned from snapshotbranch, release + # and directory entry nodes. Release node is used for testing here + query_str = """ + { + release(swhid: "swh:1:rel:ee4d20e80af850cc0f417d25dc5073792c5010d2") { + swhid + target { + ...on Directory { + swhid + } + } + } + } + """ + data, _ = utils.get_query_response(client, query_str) + assert data["release"]["target"] == { + "swhid": "swh:1:dir:0505050505050505050505050505050505050505" + } diff --git a/swh/graphql/tests/functional/test_directory_entry.py b/swh/graphql/tests/functional/test_directory_entry.py new file mode 100644 index 0000000..50a882e --- /dev/null +++ b/swh/graphql/tests/functional/test_directory_entry.py @@ -0,0 +1,37 @@ +# 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 . import utils +from ..data import get_directories + + +@pytest.mark.parametrize("directory", get_directories()) +def test_get_directory_entry_connection(client, directory): + query_str = """ + { + directory(swhid: "%s") { + swhid + entries { + nodes { + type + name { + text + } + } + } + } + } + """ + data, _ = utils.get_query_response(client, query_str % directory.swhid()) + directory_entries = data["directory"]["entries"]["nodes"] + assert len(directory_entries) == len(directory.entries) + output = [ + {"name": {"text": de.name.decode()}, "type": de.type} + for de in directory.entries + ] + for each_entry in output: + assert each_entry in directory_entries