diff --git a/swh/graphql/backends/archive.py b/swh/graphql/backends/archive.py index 7b62510..d49460b 100644 --- a/swh/graphql/backends/archive.py +++ b/swh/graphql/backends/archive.py @@ -1,84 +1,84 @@ # 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 swh.graphql import server from swh.model.swhids import ObjectType class Archive: def __init__(self): self.storage = server.get_storage() def get_origin(self, url): return self.storage.origin_get([url])[0] def get_origins(self, after=None, first=50, url_pattern=None): # STORAGE-TODO # Make them a single function in the backend if url_pattern is None: return self.storage.origin_list(page_token=after, limit=first) return self.storage.origin_search( url_pattern=url_pattern, page_token=after, limit=first ) def get_origin_visits(self, origin_url, after=None, first=50): return self.storage.origin_visit_get(origin_url, page_token=after, limit=first) def get_origin_visit(self, origin_url, visit_id): return self.storage.origin_visit_get_by(origin_url, visit_id) def get_origin_latest_visit(self, origin_url): return self.storage.origin_visit_get_latest(origin_url) def get_visit_status(self, origin_url, visit_id, after=None, first=50): return self.storage.origin_visit_status_get( origin_url, visit_id, page_token=after, limit=first ) def get_latest_visit_status(self, origin_url, visit_id): return self.storage.origin_visit_status_get_latest(origin_url, visit_id) def get_origin_snapshots(self, origin_url): return self.storage.origin_snapshot_get_all(origin_url) def get_snapshot_branches( self, snapshot, after=b"", first=50, target_types=[], name_include=None ): return self.storage.snapshot_get_branches( snapshot, branches_from=after, branches_count=first, target_types=target_types, branch_name_include_substring=name_include, ) def get_revisions(self, revision_ids): return self.storage.revision_get(revision_ids=revision_ids) def get_revision_log(self, revision_ids, after=None, first=50): return self.storage.revision_log(revisions=revision_ids, limit=first) def get_releases(self, release_ids): return self.storage.release_get(releases=release_ids) def get_directory_entries(self, directory_id, after=None, first=50): return self.storage.directory_get_entries( directory_id, limit=first, page_token=after ) def get_content(self, content_id): # FIXME, only for tests return self.storage.content_find({"sha1_git": content_id}) def is_object_available(self, object_id: str, object_type: ObjectType) -> bool: mapping = { ObjectType.CONTENT: self.storage.content_missing_per_sha1_git, ObjectType.DIRECTORY: self.storage.directory_missing, ObjectType.RELEASE: self.storage.release_missing, ObjectType.REVISION: self.storage.revision_missing, ObjectType.SNAPSHOT: self.storage.snapshot_missing, } - return not mapping[object_type]([object_id]) + return not list(mapping[object_type]([object_id])) diff --git a/swh/graphql/tests/functional/test_swhid_resolve.py b/swh/graphql/tests/functional/test_swhid_resolve.py index 3947ec8..9b0c707 100644 --- a/swh/graphql/tests/functional/test_swhid_resolve.py +++ b/swh/graphql/tests/functional/test_swhid_resolve.py @@ -1,182 +1,221 @@ # 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, get_releases, get_revisions, get_snapshots +from ..data import ( + get_contents, + get_directories, + get_releases, + get_revisions, + get_snapshots, +) def test_invalid_swhid(client): query_str = """ { resolveSwhid(swhid: "swh:1:dir:dae0d245988b472abd30a4f968b919d0019b6c7") { nodes { type } } } """ 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"] @pytest.mark.parametrize( "swhid", [ "swh:1:rel:0949d7a8c96347dba09be8d79085b8207f345412", "swh:1:rev:0949d7a8c96347dba09be8d79085b8207f345412", "swh:1:dir:0949d7a8c96347dba09be8d79085b8207f345412", "swh:1:cnt:0949d7a8c96347dba09be8d79085b8207f345412", "swh:1:snp:0949d7a8c96347dba09be8d79085b8207f345412", ], ) def test_missing_swhid(client, swhid): query_str = """ { resolveSwhid(swhid: "%s") { nodes { type } } } """ data, _ = utils.get_query_response(client, query_str % swhid) # API will return an empty list in case of a valid, non existing SWHID assert data == {"resolveSwhid": {"nodes": []}} @pytest.mark.parametrize("snapshot", get_snapshots()) def test_snapshot_swhid_resolve(client, snapshot): query_str = """ { resolveSwhid(swhid: "%s") { nodes { type target { __typename ... on Snapshot { swhid } } } } } """ data, _ = utils.get_query_response(client, query_str % snapshot.swhid()) assert data == { "resolveSwhid": { "nodes": [ { "target": { "__typename": "Snapshot", "swhid": str(snapshot.swhid()), }, "type": "snapshot", } ] } } @pytest.mark.parametrize("revision", get_revisions()) def test_revision_swhid_resolve(client, revision): query_str = """ { resolveSwhid(swhid: "%s") { nodes { type target { __typename ... on Revision { swhid } } } } } """ data, _ = utils.get_query_response(client, query_str % revision.swhid()) assert data == { "resolveSwhid": { "nodes": [ { "target": { "__typename": "Revision", "swhid": str(revision.swhid()), }, "type": "revision", } ] } } @pytest.mark.parametrize("release", get_releases()) def test_release_swhid_resolve(client, release): query_str = """ { resolveSwhid(swhid: "%s") { nodes { type target { __typename ... on Release { swhid } } } } } """ data, _ = utils.get_query_response(client, query_str % release.swhid()) assert data == { "resolveSwhid": { "nodes": [ { "target": { "__typename": "Release", "swhid": str(release.swhid()), }, "type": "release", } ] } } @pytest.mark.parametrize("directory", get_directories()) def test_directory_swhid_resolve(client, directory): query_str = """ { resolveSwhid(swhid: "%s") { nodes { type target { __typename ... on Directory { swhid } } } } } """ data, _ = utils.get_query_response(client, query_str % directory.swhid()) assert data == { "resolveSwhid": { "nodes": [ { "target": { "__typename": "Directory", "swhid": str(directory.swhid()), }, "type": "directory", } ] } } + + +@pytest.mark.parametrize("content", get_contents()) +def test_content_swhid_resolve(client, content): + query_str = """ + { + resolveSwhid(swhid: "%s") { + nodes { + type + target { + __typename + ... on Content { + swhid + } + } + } + } + } + """ + data, _ = utils.get_query_response(client, query_str % content.swhid()) + assert data == { + "resolveSwhid": { + "nodes": [ + { + "target": { + "__typename": "Content", + "swhid": str(content.swhid()), + }, + "type": "content", + } + ] + } + }