diff --git a/README.rst b/README.rst index 7a69ccdf..185ac6d1 100644 --- a/README.rst +++ b/README.rst @@ -105,6 +105,31 @@ If you want the database fixture to be automatically populated with your schema: The database will still be dropped each time. +If you've got other programmatic ways to populate the database, you would need an additional fixture, that will take care of that: + +.. code-block:: python + + @pytest.fixture(scope='function') + def db_session(postgresql): + """Session for SQLAlchemy.""" + from pyramid_fullauth.models import Base # pylint:disable=import-outside-toplevel + + # NOTE: this fstring assumes that psycopg2 >= 2.8 is used. Not sure about it's support in psycopg2cffi (PyPy) + connection = f'postgresql+psycopg2://{postgres.info.user}:@{postgres.info.host}:{postgres.info.port}/{postgres.info.dbname}' + + engine = create_engine(connection, echo=False, poolclass=NullPool) + pyramid_basemodel.Session = scoped_session(sessionmaker(extension=ZopeTransactionExtension())) + pyramid_basemodel.bind_engine( + engine, pyramid_basemodel.Session, should_create=True, should_drop=True) + + yield pyramid_basemodel.Session + + transaction.commit() + Base.metadata.drop_all(engine) + +See the original code at `pyramid_fullauth `_. +Depending on your needs, that in between code can fire alembic migrations in case of sqlalchemy stack or any other code + Connecting to already existing postgresql database diff --git a/setup.cfg b/setup.cfg index bde441da..3f9b2c04 100644 --- a/setup.cfg +++ b/setup.cfg @@ -1,5 +1,5 @@ [pycodestyle] -max-line-length = 80 +max-line-length = 100 exclude = docs/*,build/*,venv/* [pydocstyle] diff --git a/src/pytest_postgresql/compat.py b/src/pytest_postgresql/compat.py new file mode 100644 index 00000000..bd5ab8bc --- /dev/null +++ b/src/pytest_postgresql/compat.py @@ -0,0 +1,22 @@ +"""pscypog2 Compatibility module.""" +from typing import Any +from platform import python_implementation + + +try: + import psycopg2 +except ImportError: + psycopg2 = False + +# pylint:disable=unused-import +if not psycopg2: + # if there's no postgres, just go with the flow. + # pylint:disable=invalid-name + cursor = Any + connection = Any +elif python_implementation() == "PyPy": + # pylint:disable=import-error + from psycopg2cffi._impl.cursor import Cursor as cursor + from psycopg2cffi._impl.connection import Connection as connection +else: + from psycopg2._psycopg import cursor, connection # pylint:disable=no-name-in-module diff --git a/src/pytest_postgresql/factories.py b/src/pytest_postgresql/factories.py index 91935c05..fafc1908 100644 --- a/src/pytest_postgresql/factories.py +++ b/src/pytest_postgresql/factories.py @@ -21,17 +21,21 @@ import platform import subprocess from tempfile import gettempdir +from typing import List, Callable, Union, Iterable from warnings import warn import pytest +from _pytest.fixtures import FixtureRequest +from _pytest.tmpdir import TempdirFactory +from pytest_postgresql.compat import psycopg2, connection from pytest_postgresql.executor_noop import NoopExecutor -from pytest_postgresql.janitor import DatabaseJanitor, psycopg2 +from pytest_postgresql.janitor import DatabaseJanitor from pytest_postgresql.executor import PostgreSQLExecutor from pytest_postgresql.port import get_port -def get_config(request): +def get_config(request: FixtureRequest) -> dict: """Return a dictionary with config options.""" config = {} options = [ @@ -84,9 +88,11 @@ def drop_postgresql_database(user, host, port, db_name, version, password=None): def postgresql_proc( - executable=None, host=None, port=-1, user=None, password=None, - options='', startparams=None, unixsocketdir=None, logs_prefix='', -): + executable: str = None, host: str = None, port: Union[str, int, Iterable] = -1, + user: str = None, password: str = None, + options: str = '', startparams: str = None, unixsocketdir: str = None, + logs_prefix: str = '', +) -> Callable[[FixtureRequest, TempdirFactory], PostgreSQLExecutor]: """ Postgresql process factory. @@ -107,8 +113,11 @@ def postgresql_proc( :rtype: func :returns: function which makes a postgresql process """ + @pytest.fixture(scope='session') - def postgresql_proc_fixture(request, tmpdir_factory): + def postgresql_proc_fixture( + request: FixtureRequest, tmpdir_factory: TempdirFactory + ) -> PostgreSQLExecutor: """ Process fixture for PostgreSQL. @@ -161,25 +170,26 @@ def postgresql_proc_fixture(request, tmpdir_factory): def postgresql_noproc( - host=None, port=None, user=None, password=None, options='', -): + host: str = None, port: Union[str, int] = None, user: str = None, password: str = None, + options: str = '', +) -> Callable[[FixtureRequest], NoopExecutor]: """ Postgresql noprocess factory. - :param str host: hostname - :param str|int port: exact port (e.g. '8000', 8000) - :param str user: postgresql username - :param str options: Postgresql connection options - :rtype: func + :param host: hostname + :param port: exact port (e.g. '8000', 8000) + :param user: postgresql username + :param password: postgresql password + :param options: Postgresql connection options :returns: function which makes a postgresql process """ + @pytest.fixture(scope='session') - def postgresql_noproc_fixture(request): + def postgresql_noproc_fixture(request: FixtureRequest) -> NoopExecutor: """ Noop Process fixture for PostgreSQL. :param FixtureRequest request: fixture request object - :rtype: pytest_dbfixtures.executors.TCPExecutor :returns: tcp executor-like object """ config = get_config(request) @@ -202,23 +212,24 @@ def postgresql_noproc_fixture(request): return postgresql_noproc_fixture -def postgresql(process_fixture_name, db_name=None, load=None): +def postgresql( + process_fixture_name: str, db_name: str = None, load: List[str] = None +) -> Callable[[FixtureRequest], connection]: """ Return connection fixture factory for PostgreSQL. - :param str process_fixture_name: name of the process fixture - :param str db_name: database name - :param list load: SQL to automatically load into our test database - :rtype: func + :param process_fixture_name: name of the process fixture + :param db_name: database name + :param load: SQL to automatically load into our test database :returns: function which makes a connection to postgresql """ + @pytest.fixture - def postgresql_factory(request): + def postgresql_factory(request: FixtureRequest) -> connection: """ Fixture factory for PostgreSQL. :param FixtureRequest request: fixture request object - :rtype: psycopg2.connection :returns: postgresql client """ config = get_config(request) @@ -228,7 +239,8 @@ def postgresql_factory(request): 'psycopg2 or psycopg2-binary package for CPython ' 'or psycopg2cffi for Pypy.' ) - proc_fixture = request.getfixturevalue(process_fixture_name) + proc_fixture: Union[PostgreSQLExecutor, NoopExecutor] = request.getfixturevalue( + process_fixture_name) pg_host = proc_fixture.host pg_port = proc_fixture.port @@ -242,7 +254,7 @@ def postgresql_factory(request): pg_user, pg_host, pg_port, pg_db, proc_fixture.version, pg_password ): - connection = psycopg2.connect( + db_connection: connection = psycopg2.connect( dbname=pg_db, user=pg_user, password=pg_password, @@ -253,10 +265,10 @@ def postgresql_factory(request): if pg_load: for filename in pg_load: with open(filename, 'r') as _fd: - with connection.cursor() as cur: + with db_connection.cursor() as cur: cur.execute(_fd.read()) - yield connection - connection.close() + yield db_connection + db_connection.close() return postgresql_factory diff --git a/src/pytest_postgresql/janitor.py b/src/pytest_postgresql/janitor.py index 98522a39..5ec63819 100644 --- a/src/pytest_postgresql/janitor.py +++ b/src/pytest_postgresql/janitor.py @@ -1,21 +1,14 @@ """Database Janitor.""" from contextlib import contextmanager from types import TracebackType -from typing import TypeVar, Union, Optional, Type, Any +from typing import TypeVar, Union, Optional, Type from pkg_resources import parse_version + +from pytest_postgresql.compat import psycopg2, cursor + Version = type(parse_version('1')) # pylint:disable=invalid-name -try: - import psycopg2 - try: - from psycopg2._psycopg import cursor - except ImportError: - from psycopg2cffi._impl.cursor import Cursor as cursor -except ImportError: - psycopg2 = False - # if there's no postgres, just go with the flow. - cursor = Any # pylint:disable=invalid-name DatabaseJanitorType = TypeVar("DatabaseJanitorType", bound="DatabaseJanitor")