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 Counter as TCounter | |||||
| import pytest | import pytest | ||||
| from swh.core.db import BaseDb | from swh.core.db import BaseDb | ||||
| from swh.model.model import Sha1Git | |||||
| 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.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.interface import StorageInterface | ||||
| from swh.storage.postgresql.storage import Storage | from swh.storage.postgresql.storage import Storage | ||||
| @pytest.mark.parametrize( | @pytest.mark.parametrize( | ||||
| Show All 16 Lines | with BaseDb.connect(dsn).conn as conn: | ||||
| archive_api.directory_ls(directory["id"]), key=itemgetter("name") | archive_api.directory_ls(directory["id"]), key=itemgetter("name") | ||||
| ) | ) | ||||
| entries_direct = sorted( | entries_direct = sorted( | ||||
| archive_direct.directory_ls(directory["id"]), key=itemgetter("name") | archive_direct.directory_ls(directory["id"]), key=itemgetter("name") | ||||
| ) | ) | ||||
| assert entries_api == entries_direct | assert entries_api == entries_direct | ||||
| for revision in data["revision"]: | for revision in data["revision"]: | ||||
| parents_api = Counter(archive_api.revision_get_parents(revision["id"])) | parents_api: TCounter[Sha1Git] = Counter( | ||||
| parents_direct = Counter( | archive_api.revision_get_parents(revision["id"]) | ||||
| ) | |||||
| parents_direct: TCounter[Sha1Git] = Counter( | |||||
| archive_direct.revision_get_parents(revision["id"]) | archive_direct.revision_get_parents(revision["id"]) | ||||
| ) | ) | ||||
| assert parents_api == parents_direct | assert parents_api == parents_direct | ||||
| for snapshot in data["snapshot"]: | for snapshot in data["snapshot"]: | ||||
| heads_api = Counter(archive_api.snapshot_get_heads(snapshot["id"])) | heads_api: TCounter[Sha1Git] = Counter( | ||||
| heads_direct = Counter(archive_direct.snapshot_get_heads(snapshot["id"])) | archive_api.snapshot_get_heads(snapshot["id"]) | ||||
| ) | |||||
| heads_direct: TCounter[Sha1Git] = Counter( | |||||
| archive_direct.snapshot_get_heads(snapshot["id"]) | |||||
| ) | |||||
| assert heads_api == heads_direct | assert heads_api == heads_direct | ||||