PoC for using load_database by pytest-postgresql

newsfragments/721.misc.rst

@@ -0,0 +1,7 @@
+For postgresql tests use the ability of pytest-postgresql to create database
+and fill in it's contents once per full test runs.
+
+This ought to speed up the full test suite as instead of creating database
+and filling up with test data each test it'll be done only once,
+and each tests postgresql will clone the template database with data
+into a test database.

tests/conftest.py

@@ -9,6 +9,7 @@
 from zope.sqlalchemy import register
 
 from pytest_pyramid import factories
+from pytest_postgresql.factories import postgresql_proc as posgresql_proc_factory
 import pyramid_basemodel
 
 
@@ -30,11 +31,38 @@
     return request
 
 
+def load_database(**kwargs):
+    """Pre-load database with data and structure."""
+    from pyramid_fullauth.models import Base, Group  # pylint:disable=import-outside-toplevel
+
+    connection = f"postgresql+psycopg://{kwargs['user']}:@{kwargs['host']}:{kwargs['port']}/{kwargs['dbname']}"
+    engine = create_engine(connection)
+    Base.metadata.create_all(engine)
+    session = scoped_session(sessionmaker())
+    session.configure(bind=engine)
+    # This group will always be present in each test even though after each test the test database is dropped.
+    # That's because it lives in temporary database which
+    # for postgresql based tests is used to recreate a test database.
+    session.add(
+        Group(
+            name="Added in pre-init stage",
+        )
+    )
+    session.commit()
+
+
+# Createa a process fixture referencing the load_database
+postgresql_proc = posgresql_proc_factory(
+    load=[load_database],
+)
+
+
 @pytest.fixture(scope="function", params=["sqlite", "mysql", "postgresql"])
 def db_session(request):
     """Session for SQLAlchemy."""
     from pyramid_fullauth.models import Base  # pylint:disable=import-outside-toplevel
 
+    connection = ""
     if request.param == "sqlite":
         connection = "sqlite:///fullauth.sqlite"
     elif request.param == "mysql":
@@ -47,15 +75,15 @@
     engine = create_engine(connection, echo=False, poolclass=NullPool)
     pyramid_basemodel.Session = scoped_session(sessionmaker())
     register(pyramid_basemodel.Session)
-    pyramid_basemodel.bind_engine(engine, pyramid_basemodel.Session, should_create=True, should_drop=True)
+    if request.param == "postgresql":
+        pyramid_basemodel.bind_engine(engine, pyramid_basemodel.Session, should_create=False, should_drop=False)
+    else:
+        pyramid_basemodel.bind_engine(engine, pyramid_basemodel.Session, should_create=True, should_drop=True)
 
-    def destroy():
-        transaction.commit()
-        Base.metadata.drop_all(engine)
+    yield pyramid_basemodel.Session
 
-    request.addfinalizer(destroy)
-
-    return pyramid_basemodel.Session
+    transaction.commit()
+    Base.metadata.drop_all(engine)
 
 
 @pytest.fixture