Changeset View
Changeset View
Standalone View
Standalone View
swh/provenance/tests/conftest.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 | ||||
| import glob | import glob | ||||
| from os import path | from os import path | ||||
| import pytest | import pytest | ||||
| from swh.core.db.pytest_plugin import postgresql_fact | from swh.core.db.pytest_plugin import postgresql_fact | ||||
| from swh.core.utils import numfile_sortkey as sortkey | from swh.core.utils import numfile_sortkey as sortkey | ||||
| from swh.model.tests.swh_model_data import TEST_OBJECTS | from swh.model.tests.swh_model_data import TEST_OBJECTS | ||||
| import swh.provenance | import swh.provenance | ||||
| from swh.provenance.postgresql.archive import ArchivePostgreSQL | |||||
| from swh.provenance.storage.archive import ArchiveStorage | |||||
| SQL_DIR = path.join(path.dirname(swh.provenance.__file__), "sql") | SQL_DIR = path.join(path.dirname(swh.provenance.__file__), "sql") | ||||
| SQL_FILES = [ | SQL_FILES = [ | ||||
| sqlfile | sqlfile | ||||
| for sqlfile in sorted(glob.glob(path.join(SQL_DIR, "*.sql")), key=sortkey) | for sqlfile in sorted(glob.glob(path.join(SQL_DIR, "*.sql")), key=sortkey) | ||||
| if "-without-path-" not in sqlfile | if "-without-path-" not in sqlfile | ||||
| ] | ] | ||||
| Show All 27 Lines | for obj_type in ( | ||||
| "release", | "release", | ||||
| "snapshot", | "snapshot", | ||||
| "origin", | "origin", | ||||
| "origin_visit", | "origin_visit", | ||||
| "origin_visit_status", | "origin_visit_status", | ||||
| ): | ): | ||||
| getattr(swh_storage, f"{obj_type}_add")(TEST_OBJECTS[obj_type]) | getattr(swh_storage, f"{obj_type}_add")(TEST_OBJECTS[obj_type]) | ||||
| return swh_storage | return swh_storage | ||||
| @pytest.fixture | |||||
| def archive_direct(swh_storage_with_objects): | |||||
| return ArchivePostgreSQL(swh_storage_with_objects.get_db().conn) | |||||
| @pytest.fixture | |||||
| def archive_api(swh_storage_with_objects): | |||||
| return ArchiveStorage(swh_storage_with_objects) | |||||