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 | ||||
| import pytest | import pytest | ||||
| from swh.core.db import BaseDb | from swh.core.db import BaseDb | ||||
| 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.postgresql.storage import Storage | from swh.storage.postgresql.storage import Storage | ||||
| @pytest.mark.parametrize( | @pytest.mark.parametrize( | ||||
| "repo", | "repo", | ||||
| ("cmdbts2", "out-of-order", "with-merges"), | ("cmdbts2", "out-of-order", "with-merges"), | ||||
| ) | ) | ||||
| def test_archive_interface(repo: str, swh_storage: Storage) -> None: | def test_archive_interface(repo: str, swh_storage: StorageInterface) -> None: | ||||
| archive_api = ArchiveStorage(swh_storage) | archive_api = ArchiveStorage(swh_storage) | ||||
| assert isinstance(swh_storage, Storage) | |||||
| dsn = swh_storage.get_db().conn.dsn | dsn = swh_storage.get_db().conn.dsn | ||||
| with BaseDb.connect(dsn).conn as conn: | with BaseDb.connect(dsn).conn as conn: | ||||
| BaseDb.adapt_conn(conn) | BaseDb.adapt_conn(conn) | ||||
| archive_direct = ArchivePostgreSQL(conn) | archive_direct = ArchivePostgreSQL(conn) | ||||
| # read data/README.md for more details on how these datasets are generated | # read data/README.md for more details on how these datasets are generated | ||||
| data = load_repo_data(repo) | data = load_repo_data(repo) | ||||
| fill_storage(swh_storage, data) | fill_storage(swh_storage, data) | ||||
| Show All 20 Lines | |||||