Differential D3452 Diff 12227 swh/storage/pytest_plugin.py

Changeset View

Standalone View

swh/storage/pytest_plugin.py

This file was copied from swh/storage/tests/conftest.py.

# Copyright (C) 2019 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 glob		import glob
import pytest
import multiprocessing.util
from typing import Union

from os import path, environ		from os import path, environ
from hypothesis import settings		from typing import Union
from typing import Dict

try:		import pytest
import pytest_cov.embed
except ImportError:		import swh.storage
pytest_cov = None

from pytest_postgresql import factories		from pytest_postgresql import factories
from pytest_postgresql.janitor import DatabaseJanitor, psycopg2, Version		from pytest_postgresql.janitor import DatabaseJanitor, psycopg2, Version

import swh.storage

from swh.core.utils import numfile_sortkey as sortkey		from swh.core.utils import numfile_sortkey as sortkey
from swh.model.tests.generate_testdata import gen_contents, gen_origins		from swh.storage import get_storage
from swh.model.model import (
Content,
Directory,
Origin,
OriginVisit,
Release,
Revision,
SkippedContent,
Snapshot,
)


OBJECT_FACTORY = {
"content": Content.from_dict,
"directory": Directory.from_dict,
"origin": Origin.from_dict,
"origin_visit": OriginVisit.from_dict,
"release": Release.from_dict,
"revision": Revision.from_dict,
"skipped_content": SkippedContent.from_dict,
"snapshot": Snapshot.from_dict,
}

SQL_DIR = path.join(path.dirname(swh.storage.__file__), "sql")		SQL_DIR = path.join(path.dirname(swh.storage.__file__), "sql")

environ["LC_ALL"] = "C.UTF-8"		environ["LC_ALL"] = "C.UTF-8"

DUMP_FILES = path.join(SQL_DIR, "*.sql")		DUMP_FILES = path.join(SQL_DIR, "*.sql")

# 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		@pytest.fixture
def swh_storage_backend_config(postgresql_proc, swh_storage_postgresql):		def swh_storage_backend_config(postgresql_proc, swh_storage_postgresql):
yield {		yield {
"cls": "local",		"cls": "local",
"db": "postgresql://{user}@{host}:{port}/{dbname}".format(		"db": "postgresql://{user}@{host}:{port}/{dbname}".format(
host=postgresql_proc.host,		host=postgresql_proc.host,
port=postgresql_proc.port,		port=postgresql_proc.port,
user="postgres",		user="postgres",
dbname="tests",		dbname="tests",
),		),
"objstorage": {"cls": "memory", "args": {}},		"objstorage": {"cls": "memory", "args": {}},
"journal_writer": {"cls": "memory",},		"journal_writer": {"cls": "memory",},
}		}


@pytest.fixture		@pytest.fixture
def swh_storage(swh_storage_backend_config):		def swh_storage(swh_storage_backend_config):
return swh.storage.get_storage(cls="validate", storage=swh_storage_backend_config)		return get_storage(cls="validate", storage=swh_storage_backend_config)


@pytest.fixture
def swh_contents(swh_storage):
contents = 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):
origins = gen_origins(n=100)
swh_storage.origin_add(origins)
return origins


# the postgres_fact factory fixture below is mostly a copy of the code		# the postgres_fact factory fixture below is mostly a copy of the code
# from pytest-postgresql. We need a custom version here to be able to		# from pytest-postgresql. We need a custom version here to be able to
# specify our version of the DBJanitor we use.		# specify our version of the DBJanitor we use.
def postgresql_fact(process_fixture_name, db_name=None, dump_files=DUMP_FILES):		def postgresql_fact(process_fixture_name, db_name=None, dump_files=DUMP_FILES):
@pytest.fixture		@pytest.fixture
def postgresql_factory(request):		def postgresql_factory(request):
▲ Show 20 Lines • Show All 120 Lines • ▼ Show 20 Lines	def drop(self):
(self.db_name,),		(self.db_name,),
)		)
cur.execute(		cur.execute(
"SELECT pg_terminate_backend(pg_stat_activity.{})"		"SELECT pg_terminate_backend(pg_stat_activity.{})"
"FROM pg_stat_activity "		"FROM pg_stat_activity "
"WHERE pg_stat_activity.datname = %s;".format(pid_column),		"WHERE pg_stat_activity.datname = %s;".format(pid_column),
(self.db_name,),		(self.db_name,),
)		)


@pytest.fixture
def sample_data() -> Dict:
"""Pre-defined sample storage object data to manipulate

Returns:
Dict of data (keys: content, directory, revision, release, person,
origin)

"""
from .storage_data import data

return {
"content": [data.cont, data.cont2],
"content_metadata": [data.cont3],
"skipped_content": [data.skipped_cont, data.skipped_cont2],
"person": [data.person],
"directory": [data.dir2, data.dir],
"revision": [data.revision, data.revision2, data.revision3],
"release": [data.release, data.release2, data.release3],
"snapshot": [data.snapshot],
"origin": [data.origin, data.origin2],
"fetcher": [data.metadata_fetcher],
"authority": [data.metadata_authority],
"origin_metadata": [data.origin_metadata, data.origin_metadata2],
}