Tell warehouse.db to reuse db_session pytest fixture

tests/common/db/accounts.py

@@ -22,6 +22,7 @@
 class UserFactory(WarehouseFactory):
     class Meta:
         model = User
+        # sqlalchemy_session_persistence = 'commit'
 
     username = factory.Faker("pystr", max_chars=12)
     name = factory.Faker("word")

tests/conftest.py

@@ -25,7 +25,9 @@
 import pyramid.testing
 import pytest
 import stripe
+import transaction
 import webtest as _webtest
+import zope.sqlalchemy
 
 from jinja2 import Environment, FileSystemLoader
 from psycopg.errors import InvalidCatalogName
@@ -622,18 +624,39 @@ 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 tm():
+    tm = transaction.TransactionManager(explicit=True)
+    tm.begin()
+    tm.doom()
 
+    yield tm
+
+    tm.abort()
+
+
+@pytest.fixture
+def webtest(app_config, tm, db_session):
     # We want to disable anything that relies on TLS here.
     app_config.add_settings(enforce_https=False)
 
-    try:
-        yield _TestApp(app_config.make_wsgi_app())
-    finally:
-        app_config.registry["sqlalchemy.engine"].dispose()
+    # Register the transaction manager with the db_session
+    zope.sqlalchemy.register(db_session, transaction_manager=tm)
+
+    # Create WSGI app with current test settings
+    # (this could be a fixture)
+    app = app_config.make_wsgi_app()
+
+    # Register the app with the external test environment, telling
+    # warehouse.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
 
 
 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

tests/unit/test_db.py

@@ -120,6 +120,7 @@ def raiser():
     request = pretend.stub(
         find_service=pyramid_services.find_service,
         registry={"sqlalchemy.engine": engine},
+        environ={},
     )
 
     with pytest.raises(DatabaseNotAvailableError):
@@ -151,6 +152,7 @@ def test_create_session(monkeypatch, pyramid_services):
         registry={"sqlalchemy.engine": engine},
         tm=pretend.stub(),
         add_finished_callback=pretend.call_recorder(lambda callback: None),
+        environ={},
     )
 
     request2 = pretend.stub()
@@ -212,6 +214,7 @@ def test_create_session_read_only_mode(
         tm=pretend.stub(doom=pretend.call_recorder(lambda: None)),
         add_finished_callback=lambda callback: None,
         user=pretend.stub(is_superuser=is_superuser),
+        environ={},
     )
 
     assert _create_session(request) is session_obj

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")