Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/conftest.py
# Copyright (C) 2019-2020 The Software Heritage developers | # Copyright (C) 2019-2020 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 pytest | import pytest | ||||
import multiprocessing.util | import multiprocessing.util | ||||
from hypothesis import settings | from hypothesis import settings | ||||
try: | try: | ||||
import pytest_cov.embed | import pytest_cov.embed | ||||
except ImportError: | except ImportError: | ||||
pytest_cov = None | pytest_cov = None | ||||
from typing import Iterable | |||||
from swh.model.model import BaseContent | |||||
from swh.model.tests.generate_testdata import gen_contents, gen_origins | from swh.model.tests.generate_testdata import gen_contents, gen_origins | ||||
from swh.storage import get_storage | from swh.storage import get_storage | ||||
from swh.storage.interface import StorageInterface | |||||
# define tests profile. Full documentation is at: | # define tests profile. Full documentation is at: | ||||
# https://hypothesis.readthedocs.io/en/latest/settings.html#settings-profiles | # https://hypothesis.readthedocs.io/en/latest/settings.html#settings-profiles | ||||
settings.register_profile("fast", max_examples=5, deadline=5000) | settings.register_profile("fast", max_examples=5, deadline=5000) | ||||
settings.register_profile("slow", max_examples=20, deadline=5000) | settings.register_profile("slow", max_examples=20, deadline=5000) | ||||
if pytest_cov is not None: | if pytest_cov is not None: | ||||
Show All 20 Lines | |||||
@pytest.fixture | @pytest.fixture | ||||
def swh_storage(swh_storage_backend_config): | def swh_storage(swh_storage_backend_config): | ||||
return get_storage(cls="validate", storage=swh_storage_backend_config) | return get_storage(cls="validate", storage=swh_storage_backend_config) | ||||
@pytest.fixture | @pytest.fixture | ||||
def swh_contents(swh_storage): | def swh_contents(swh_storage: StorageInterface) -> Iterable[BaseContent]: | ||||
contents = gen_contents(n=20) | contents = [BaseContent.from_dict(c) for c in gen_contents(n=20)] | ||||
swh_storage.content_add([c for c in contents if c["status"] != "absent"]) | swh_storage.content_add([c for c in contents if c.status != "absent"]) | ||||
swh_storage.skipped_content_add([c for c in contents if c["status"] == "absent"]) | swh_storage.skipped_content_add([c for c in contents if c.status == "absent"]) | ||||
return contents | return contents | ||||
@pytest.fixture | @pytest.fixture | ||||
def swh_origins(swh_storage): | def swh_origins(swh_storage): | ||||
origins = gen_origins(n=100) | origins = gen_origins(n=100) | ||||
swh_storage.origin_add(origins) | swh_storage.origin_add(origins) | ||||
return origins | return origins |