Changeset View
Changeset View
Standalone View
Standalone View
swh/provenance/tests/conftest.py
| Show All 10 Lines | |||||
| import pytest | import pytest | ||||
| from typing_extensions import TypedDict | from typing_extensions import TypedDict | ||||
| from swh.core.db import BaseDb | from swh.core.db import BaseDb | ||||
| from swh.journal.serializers import msgpack_ext_hook | from swh.journal.serializers import msgpack_ext_hook | ||||
| from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import hash_to_bytes | ||||
| from swh.model.model import Sha1Git | from swh.model.model import Sha1Git | ||||
| from swh.model.tests.swh_model_data import TEST_OBJECTS | from swh.model.tests.swh_model_data import TEST_OBJECTS | ||||
| from swh.provenance import get_provenance | |||||
| 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.storage.replay import process_replay_objects | from swh.storage.replay import process_replay_objects | ||||
| @pytest.fixture(params=["with-path", "without-path"]) | @pytest.fixture(params=["with-path", "without-path"]) | ||||
| def provenance(request, postgresql): | def provenance(request, postgresql): | ||||
| """return a working and initialized provenance db""" | """return a working and initialized provenance db""" | ||||
| from swh.core.cli.db import populate_database_for_package | from swh.core.cli.db import populate_database_for_package | ||||
| flavor = request.param | flavor = request.param | ||||
| populate_database_for_package("swh.provenance", postgresql.dsn, flavor=flavor) | populate_database_for_package("swh.provenance", postgresql.dsn, flavor=flavor) | ||||
| from swh.provenance.backend import ProvenanceBackend | |||||
| BaseDb.adapt_conn(postgresql) | BaseDb.adapt_conn(postgresql) | ||||
| prov = ProvenanceBackend(postgresql) | |||||
| args = dict(tuple(item.split("=")) for item in postgresql.dsn.split()) | |||||
| args.pop("options") | |||||
| prov = get_provenance(cls="local", db=args) | |||||
| assert prov.storage.flavor == flavor | assert prov.storage.flavor == flavor | ||||
| # in test sessions, we DO want to raise any exception occurring at commit time | # in test sessions, we DO want to raise any exception occurring at commit time | ||||
| prov.raise_on_commit = True | prov.storage.raise_on_commit = True | ||||
| return prov | return prov | ||||
| @pytest.fixture | @pytest.fixture | ||||
| def swh_storage_with_objects(swh_storage): | def swh_storage_with_objects(swh_storage): | ||||
| """return a Storage object (postgresql-based by default) with a few of each | """return a Storage object (postgresql-based by default) with a few of each | ||||
| object type in it | object type in it | ||||
| ▲ Show 20 Lines • Show All 189 Lines • Show Last 20 Lines | |||||