Differential D7934 Diff 28586 swh/core/db/tests/test_db.py

Changeset View

Standalone View

swh/core/db/tests/test_db.py

	Show All 10 Lines
	from typing import Any			from typing import Any
	from unittest.mock import MagicMock, Mock			from unittest.mock import MagicMock, Mock
	import uuid			import uuid

	from hypothesis import given, settings, strategies			from hypothesis import given, settings, strategies
	from hypothesis.extra.pytz import timezones			from hypothesis.extra.pytz import timezones
	import psycopg2			import psycopg2
	import pytest			import pytest
				from pytest_postgresql import factories
	from typing_extensions import Protocol			from typing_extensions import Protocol

	from swh.core.db import BaseDb			from swh.core.db import BaseDb
	from swh.core.db.common import db_transaction, db_transaction_generator			from swh.core.db.common import db_transaction, db_transaction_generator
	from swh.core.db.pytest_plugin import postgresql_fact
	from swh.core.db.tests.conftest import function_scoped_fixture_check			from swh.core.db.tests.conftest import function_scoped_fixture_check


	# workaround mypy bug https://github.com/python/mypy/issues/5485			# workaround mypy bug https://github.com/python/mypy/issues/5485
	class Converter(Protocol):			class Converter(Protocol):
	def __call__(self, x: Any) -> Any:			def __call__(self, x: Any) -> Any:
	...			...

	▲ Show 20 Lines • Show All 207 Lines • ▼ Show 20 Lines


	def convert_lines(cur):			def convert_lines(cur):
	return [			return [
	tuple(field.out_converter(x) for x, field in zip(line, FIELDS)) for line in cur			tuple(field.out_converter(x) for x, field in zip(line, FIELDS)) for line in cur
	]			]


	test_db = postgresql_fact("postgresql_proc", dbname="test-db2")			test_db = factories.postgresql("postgresql_proc", dbname="test-db2")


	@pytest.fixture			@pytest.fixture
	def db_with_data(test_db, request):			def db_with_data(test_db, request):
	"""Fixture to initialize a db with some data out of the "INIT_SQL above"""			"""Fixture to initialize a db with some data out of the "INIT_SQL above"""
	db = BaseDb.connect(test_db.dsn)			db = BaseDb.connect(test_db.dsn)
	with db.cursor() as cur:			with db.cursor() as cur:
	psycopg2.extras.register_default_jsonb(cur)			psycopg2.extras.register_default_jsonb(cur)
	▲ Show 20 Lines • Show All 211 Lines • Show Last 20 Lines