Tell warehouse.db to reuse db_session pytest fixture (#16031)

Co-authored-by: Mike Fiedler <miketheman@gmail.com>
Co-authored-by: Jeff Zohrab <jzohrab@gmail.com>

Author: 3 people

tests/conftest.py

@@ -25,6 +25,7 @@
 import pyramid.testing
 import pytest
 import stripe
+import transaction
 import webtest as _webtest
 
 from jinja2 import Environment, FileSystemLoader
@@ -298,8 +299,7 @@ def mock_manifest_cache_buster():
     return MockManifestCacheBuster
 
 
-@pytest.fixture(scope="session")
-def app_config(database):
+def get_app_config(database, nondefaults=None):
     settings = {
         "warehouse.prevent_esi": True,
         "warehouse.token": "insecure token",
@@ -329,20 +329,42 @@ def app_config(database):
         "statuspage.url": "https://2p66nmmycsj3.statuspage.io",
         "warehouse.xmlrpc.cache.url": "redis://localhost:0/",
     }
+
+    if nondefaults:
+        settings.update(nondefaults)
+
     with mock.patch.object(config, "ManifestCacheBuster", MockManifestCacheBuster):
         with mock.patch("warehouse.admin.ManifestCacheBuster", MockManifestCacheBuster):
             with mock.patch.object(static, "whitenoise_add_manifest"):
                 cfg = config.configure(settings=settings)
 
-    # Ensure our migrations have been ran.
+    # Run migrations:
+    # This might harmlessly run multiple times if there are several app config fixtures
+    # in the test session, using the same database.
     alembic.command.upgrade(cfg.alembic_config(), "head")
 
     return cfg
 
 
-@pytest.fixture
-def db_session(app_config):
-    engine = app_config.registry["sqlalchemy.engine"]
+@contextmanager
+def get_db_session_for_app_config(app_config):
+    """
+    Refactor: This helper function is designed to help fixtures yield a database
+    session for a particular app_config.
+
+    It needs the app_config in order to fetch the database engine that's owned
+    by the config.
+    """
+
+    # TODO: We possibly accept 2 instances of the sqlalchemy engine.
+    # There's a bit of circular dependencies in place:
+    # 1) To create a database session, we need to create an app config registry
+    #    and read config.registry["sqlalchemy.engine"]
+    # 2) To create an app config registry, we need to be able to dictate the
+    #    database session through the initial config.
+    #
+    # 1) and 2) clash.
+    engine = app_config.registry["sqlalchemy.engine"]  # get_sqlalchemy_engine(database)
     conn = engine.connect()
     trans = conn.begin()
     session = Session(bind=conn, join_transaction_mode="create_savepoint")
@@ -357,6 +379,35 @@ def db_session(app_config):
         engine.dispose()
 
 
+@pytest.fixture(scope="session")
+def app_config(database):
+
+    return get_app_config(database)
+
+
+@pytest.fixture(scope="session")
+def app_config_dbsession_from_env(database):
+
+    nondefaults = {
+        "warehouse.db_create_session": lambda r: r.environ.get("warehouse.db_session")
+    }
+
+    return get_app_config(database, nondefaults)
+
+
+@pytest.fixture
+def db_session(app_config):
+    """
+    Refactor:
+
+    This fixture actually manages a specific app_config paired with a database
+    connection. For this reason, it's suggested to change the name to
+    db_and_app, and yield both app_config and db_session.
+    """
+    with get_db_session_for_app_config(app_config) as _db_session:
+        yield _db_session
+
+
 @pytest.fixture
 def user_service(db_session, metrics, remote_addr):
     return account_services.DatabaseUserService(
@@ -622,18 +673,42 @@ def xmlrpc(self, path, method, *args):
 
 
 @pytest.fixture
-def webtest(app_config):
-    # TODO: Ensure that we have per test isolation of the database level
-    #       changes. This probably involves flushing the database or something
-    #       between test cases to wipe any committed changes.
+def webtest(app_config_dbsession_from_env):
+    """
+    This fixture yields a test app with an alternative Pyramid configuration,
+    injecting the database session and transaction manager into the app.
+
+    This is because the Warehouse app normally manages its own database session.
+
+    After the fixture has yielded the app, the transaction is rolled back and
+    the database is left in its previous state.
+    """
 
     # We want to disable anything that relies on TLS here.
-    app_config.add_settings(enforce_https=False)
+    app_config_dbsession_from_env.add_settings(enforce_https=False)
+
+    app = app_config_dbsession_from_env.make_wsgi_app()
+
+    # Create a new transaction manager for dependant test cases
+    tm = transaction.TransactionManager(explicit=True)
+    tm.begin()
+    tm.doom()
+
+    with get_db_session_for_app_config(app_config_dbsession_from_env) as _db_session:
+        # Register the app with the external test environment, telling
+        # request.db to use this db_session and use the Transaction manager.
+        testapp = _TestApp(
+            app,
+            extra_environ={
+                "warehouse.db_session": _db_session,
+                "tm.active": True,  # disable pyramid_tm
+                "tm.manager": tm,  # pass in our own tm for the app to use
+            },
+        )
+        yield testapp
 
-    try:
-        yield _TestApp(app_config.make_wsgi_app())
-    finally:
-        app_config.registry["sqlalchemy.engine"].dispose()
+    # Abort the transaction, leaving database in previous state
+    tm.abort()
 
 
 class _MockRedis:

tests/functional/test_user_profile.py

@@ -0,0 +1,30 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from tests.common.db.accounts import UserFactory
+
+
+def test_user_profile(webtest):
+    """
+    This test is maintained as a POC for future tests that want to add data to
+    the database and test HTTP endpoints afterwards.
+
+    The trick is to use the ``webtest`` fixture which will create a special
+    instance of the Warehouse WSGI app, sharing the same DB session as is active
+    in pytest.
+    """
+    # Create a user
+    user = UserFactory.create()
+    assert user.username
+    # ...and verify that the user's profile page exists
+    resp = webtest.get(f"/user/{user.username}/")
+    assert resp.status_code == 200

warehouse/db.py

@@ -187,8 +187,12 @@ def includeme(config):
         pool_timeout=20,
     )
 
-    # Register our request.db property
-    config.add_request_method(_create_session, name="db", reify=True)
+    # Possibly override how to fetch new db sessions from config.settings
+    #  Useful in test fixtures
+    db_session_factory = config.registry.settings.get(
+        "warehouse.db_create_session", _create_session
+    )
+    config.add_request_method(db_session_factory, name="db", reify=True)
 
     # Set a custom JSON serializer for psycopg
     renderer = JSON()