Tell warehouse.db to reuse db_session test fixture

tests/conftest.py

@@ -27,6 +27,7 @@
 import stripe
 import transaction
 import webtest as _webtest
+import zope.sqlalchemy
 
 from jinja2 import Environment, FileSystemLoader
 from psycopg.errors import InvalidCatalogName
@@ -623,33 +624,36 @@ def xmlrpc(self, path, method, *args):
 
 
 @pytest.fixture
-def webtest(app_config, db_session):
+def tm():
+    tm = transaction.TransactionManager(explicit=True)
+    tm.begin()
+    tm.doom()
+
+    yield tm
+
+    tm.abort()
+
+
+@pytest.fixture
+def webtest(app_config, tm, db_session):
     # 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.
 
     # We want to disable anything that relies on TLS here.
     app_config.add_settings(enforce_https=False)
-
-    try:
-        tm = transaction.TransactionManager(explicit=True)
-        tm.begin()
-        # tm.doom()  # ensure no one can call tm.commit() manually
-
-        # I tried setting the tm.manager_hook to return the given
-        # transaction, but it doesn't appear to be called.  e.g.,
-        # if I change it to return "42", the test still runs.
-        app_config.add_settings({ "tm.manager_hook": lambda request: tm })
-
-        app = app_config.make_wsgi_app()
-        testapp = _TestApp(app, extra_environ={
-            'tm.active': True,    # disable pyramid_tm
-            'tm.manager': tm,    # pass in our own tm for the app to use
-        })
-        yield testapp
-    finally:
-        tm.abort()
-        app_config.registry["sqlalchemy.engine"].dispose()
+    zope.sqlalchemy.register(db_session, transaction_manager=tm)
+
+    app = app_config.make_wsgi_app()
+    testapp = _TestApp(
+        app,
+        extra_environ={
+            "warehouse.db_session": db_session,  # Tell warehouse.db to use this dbsession
+            "tm.active": True,  # disable pyramid_tm
+            "tm.manager": tm,  # pass in our own tm for the app to use
+        },
+    )
+    yield testapp
 
 
 class _MockRedis:

tests/functional/test_user_profile.py

@@ -1,13 +1,27 @@
-from tests.common.db.accounts import UserFactory
+# 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 sqlalchemy import text
 
-def test_user_profile(db_session, webtest):
+from tests.common.db.accounts import UserFactory
+
+
+def test_user_profile(webtest, db_session):
     user = UserFactory.create()
     assert user.username
     print(f"got user {user.username}", flush=True)
     result = db_session.execute(text("select * from users"))
     actual = ["; ".join([f"{s}" for s in row]) for row in result]
     print(actual, flush=True)
-    # vist user's page
+    # visit user's page
     resp = webtest.get(f"/user/{user.username}/")
     assert resp.status_code == 200

warehouse/db.py

@@ -134,6 +134,12 @@ def _configure_alembic(config):
 
 
 def _create_session(request):
+
+    # Use pre-configured db session (usually for integration test transactions)
+    db_session = request.environ.get("warehouse.db_session")
+    if db_session:
+        return db_session
+
     metrics = request.find_service(IMetricsService, context=None)
     metrics.increment("warehouse.db.session.start")