Differential D8214 Diff 29613 swh/indexer/tests/conftest.py

Changeset View

Standalone View

swh/indexer/tests/conftest.py

	# Copyright (C) 2019-2022 The Software Heritage developers			# Copyright (C) 2019-2022 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

	from datetime import timedelta			from datetime import timedelta
	from functools import partial			from functools import partial
	import os			import os
	from typing import List, Tuple			from typing import List, Tuple
	from unittest.mock import patch			from unittest.mock import patch

	import pytest			import pytest
	from pytest_postgresql import factories			from pytest_postgresql import factories
				import sentry_sdk
	import yaml			import yaml

	from swh.core.db.pytest_plugin import initialize_database_for_module			from swh.core.db.pytest_plugin import initialize_database_for_module
	from swh.indexer.storage import IndexerStorage, get_indexer_storage			from swh.indexer.storage import IndexerStorage, get_indexer_storage
	from swh.objstorage.factory import get_objstorage			from swh.objstorage.factory import get_objstorage
	from swh.storage import get_storage			from swh.storage import get_storage

	from .utils import fill_obj_storage, fill_storage			from .utils import fill_obj_storage, fill_storage
	▲ Show 20 Lines • Show All 103 Lines • ▼ Show 20 Lines

	@pytest.fixture			@pytest.fixture
	def swh_config(swh_indexer_config, monkeypatch, tmp_path):			def swh_config(swh_indexer_config, monkeypatch, tmp_path):
	conffile = os.path.join(str(tmp_path), "indexer.yml")			conffile = os.path.join(str(tmp_path), "indexer.yml")
	with open(conffile, "w") as f:			with open(conffile, "w") as f:
	f.write(yaml.dump(swh_indexer_config))			f.write(yaml.dump(swh_indexer_config))
	monkeypatch.setenv("SWH_CONFIG_FILENAME", conffile)			monkeypatch.setenv("SWH_CONFIG_FILENAME", conffile)
	return conffile			return conffile


				@pytest.fixture
				def sentry_init():
				# Inspired by
				# https://github.com/getsentry/sentry-python/blob/1.5.9/tests/conftest.py#L168-L184

				def inner(a, *kw):
				hub = sentry_sdk.Hub.current
				client = sentry_sdk.Client(a, *kw)
				hub.bind_client(client)
				client.transport = TestTransport()

				class TestTransport:
				def __init__(self):
				self.events = []
				self.envelopes = []

				def capture_event(self, event):
				self.events.append(event)

				def capture_envelope(self, envelope):
				self.envelopes.append(envelope)

				with sentry_sdk.Hub(None):
				yield inner


				@pytest.fixture
				def sentry_capture_events(monkeypatch):
				# Copied from
				# https://github.com/getsentry/sentry-python/blob/1.5.9/tests/conftest.py#L195-L217
				def inner():
				events = []
				test_client = sentry_sdk.Hub.current.client
				old_capture_event = test_client.transport.capture_event
				old_capture_envelope = test_client.transport.capture_envelope

				def append_event(event):
				events.append(event)
				return old_capture_event(event)

				def append_envelope(envelope):
				for item in envelope:
				if item.headers.get("type") in ("event", "transaction"):
				test_client.transport.capture_event(item.payload.json)
				return old_capture_envelope(envelope)

				monkeypatch.setattr(test_client.transport, "capture_event", append_event)
				monkeypatch.setattr(test_client.transport, "capture_envelope", append_envelope)
				return events

				return inner