Changeset View
Changeset View
Standalone View
Standalone View
swh/provenance/tests/test_archive_interface.py
| # Copyright (C) 2021 The Software Heritage developers | # Copyright (C) 2021 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| from collections import Counter | from collections import Counter | ||||
| from operator import itemgetter | from operator import itemgetter | ||||
| from typing import Any | |||||
| from typing import Counter as TCounter | from typing import Counter as TCounter | ||||
| from typing import Dict, List, Set, Tuple, Type, Union | from typing import Dict, Iterable, List, Set, Tuple, Type, Union | ||||
| import pytest | import pytest | ||||
| from swh.core.db import BaseDb | from swh.core.db import BaseDb | ||||
| from swh.graph.naive_client import NaiveClient | from swh.graph.naive_client import NaiveClient | ||||
| from swh.model.model import ( | from swh.model.model import ( | ||||
| BaseModel, | BaseModel, | ||||
| Content, | Content, | ||||
| Directory, | Directory, | ||||
| DirectoryEntry, | DirectoryEntry, | ||||
| Origin, | Origin, | ||||
| OriginVisit, | OriginVisit, | ||||
| OriginVisitStatus, | OriginVisitStatus, | ||||
| Revision, | Revision, | ||||
| Sha1Git, | Sha1Git, | ||||
| Snapshot, | Snapshot, | ||||
| SnapshotBranch, | SnapshotBranch, | ||||
| TargetType, | TargetType, | ||||
| ) | ) | ||||
| from swh.model.swhids import CoreSWHID, ExtendedObjectType, ExtendedSWHID | from swh.model.swhids import CoreSWHID, ExtendedObjectType, ExtendedSWHID | ||||
| from swh.provenance.archive import ArchiveInterface | from swh.provenance.archive import ArchiveInterface | ||||
| from swh.provenance.multiplexer.archive import ArchiveMultiplexed | |||||
| from swh.provenance.postgresql.archive import ArchivePostgreSQL | from swh.provenance.postgresql.archive import ArchivePostgreSQL | ||||
| from swh.provenance.storage.archive import ArchiveStorage | from swh.provenance.storage.archive import ArchiveStorage | ||||
| from swh.provenance.swhgraph.archive import ArchiveGraph | from swh.provenance.swhgraph.archive import ArchiveGraph | ||||
| from swh.provenance.tests.conftest import fill_storage, load_repo_data | from swh.provenance.tests.conftest import fill_storage, load_repo_data | ||||
| from swh.storage.interface import StorageInterface | |||||
| from swh.storage.postgresql.storage import Storage | from swh.storage.postgresql.storage import Storage | ||||
| class ArchiveNoop: | |||||
| storage: StorageInterface | |||||
| def directory_ls(self, id: Sha1Git, minsize: int = 0) -> Iterable[Dict[str, Any]]: | |||||
| return [] | |||||
| def revision_get_parents(self, id: Sha1Git) -> Iterable[Sha1Git]: | |||||
| return [] | |||||
| def snapshot_get_heads(self, id: Sha1Git) -> Iterable[Sha1Git]: | |||||
| return [] | |||||
| def check_directory_ls( | def check_directory_ls( | ||||
| reference: ArchiveInterface, archive: ArchiveInterface, data: Dict[str, List[dict]] | reference: ArchiveInterface, archive: ArchiveInterface, data: Dict[str, List[dict]] | ||||
| ) -> None: | ) -> None: | ||||
| for directory in data["directory"]: | for directory in data["directory"]: | ||||
| entries_ref = sorted( | entries_ref = sorted( | ||||
| reference.directory_ls(directory["id"]), key=itemgetter("name") | reference.directory_ls(directory["id"]), key=itemgetter("name") | ||||
| ) | ) | ||||
| entries = sorted(archive.directory_ls(directory["id"]), key=itemgetter("name")) | entries = sorted(archive.directory_ls(directory["id"]), key=itemgetter("name")) | ||||
| ▲ Show 20 Lines • Show All 163 Lines • ▼ Show 20 Lines | def test_archive_interface(repo: str, archive: ArchiveInterface) -> None: | ||||
| # test against ArchiveGraph | # test against ArchiveGraph | ||||
| nodes, edges = get_graph_data(data) | nodes, edges = get_graph_data(data) | ||||
| graph = NaiveClient(nodes=nodes, edges=edges) | graph = NaiveClient(nodes=nodes, edges=edges) | ||||
| archive_graph = ArchiveGraph(graph, archive.storage) | archive_graph = ArchiveGraph(graph, archive.storage) | ||||
| with pytest.raises(NotImplementedError): | with pytest.raises(NotImplementedError): | ||||
| check_directory_ls(archive, archive_graph, data) | check_directory_ls(archive, archive_graph, data) | ||||
| check_revision_get_parents(archive, archive_graph, data) | check_revision_get_parents(archive, archive_graph, data) | ||||
| check_snapshot_get_heads(archive, archive_graph, data) | check_snapshot_get_heads(archive, archive_graph, data) | ||||
| # test against ArchiveMultiplexer | |||||
| archive_multiplexed = ArchiveMultiplexed( | |||||
| [ArchiveNoop(), archive_graph, archive_api] | |||||
| ) | |||||
| check_directory_ls(archive, archive_multiplexed, data) | |||||
| check_revision_get_parents(archive, archive_multiplexed, data) | |||||
| check_snapshot_get_heads(archive, archive_multiplexed, data) | |||||
| def test_noop_multiplexer(): | |||||
| archive = ArchiveMultiplexed([ArchiveNoop()]) | |||||
| assert not archive.directory_ls(Sha1Git(b"abcd")) | |||||
| assert not archive.revision_get_parents(Sha1Git(b"abcd")) | |||||
| assert not archive.snapshot_get_heads(Sha1Git(b"abcd")) | |||||