conftest.py
# Copyright (C) 2019-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
import pytest
import multiprocessing.util
from hypothesis import settings
try:
import pytest_cov.embed
except ImportError:
pytest_cov = None
from typing import Iterable
from swh.model.model import BaseContent, Origin
from swh.model.tests.generate_testdata import gen_contents, gen_origins
from swh.storage.interface import StorageInterface
# define tests profile. Full documentation is at:
# https://hypothesis.readthedocs.io/en/latest/settings.html#settings-profiles
settings.register_profile("fast", max_examples=5, deadline=5000)
settings.register_profile("slow", max_examples=20, deadline=5000)
if pytest_cov is not None:
# pytest_cov + multiprocessing can cause a segmentation fault when starting
# the child process <https://forge.softwareheritage.org/P706>; so we're
# removing pytest-coverage's hook that runs when a child process starts.
# This means code run in child processes won't be counted in the coverage
# report, but this is not an issue because the only code that runs only in
# child processes is the RPC server.
for (key, value) in multiprocessing.util._afterfork_registry.items():
if value is pytest_cov.embed.multiprocessing_start:
del multiprocessing.util._afterfork_registry[key]
break
else:
assert False, "missing pytest_cov.embed.multiprocessing_start?"
@pytest.fixture
def swh_storage_backend_config(swh_storage_backend_config):
"""storage should test with its journal writer collaborator on
"""
yield {**swh_storage_backend_config, "journal_writer": {"cls": "memory",}}
@pytest.fixture
def swh_contents(swh_storage: StorageInterface) -> Iterable[BaseContent]:
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.skipped_content_add([c for c in contents if c.status == "absent"])
return contents
@pytest.fixture
def swh_origins(swh_storage: StorageInterface) -> Iterable[Origin]:
origins = [Origin.from_dict(o) for o in gen_origins(n=100)]
swh_storage.origin_add(origins)
return origins