Merge pull request #6 from charness/master

Expire session state as at the top level

Author: jeancochrane

pytest_flask_sqlalchemy/fixtures.py

@@ -53,6 +53,10 @@ def _transaction(request, _db, mocker):
     @sa.event.listens_for(session, 'after_transaction_end')
     def restart_savepoint(session, trans):
         if trans.nested and not trans._parent.nested:
+            # ensure that state is expired the way
+            # session.commit() at the top level normally does
+            session.expire_all()
+
             session.begin_nested()
 
     # Force the connection to use nested transactions

tests/_conftest.py

@@ -79,3 +79,35 @@ def drop_tables():
         _db.drop_all()
 
     return Person
+
+
+@pytest.fixture(scope='module')
+def account_address(request, _db, person):
+    '''
+    Create tables to use for testing deletes and relationships.
+    '''
+    class Account(_db.Model):
+        __tablename__ = 'account'
+
+        id = _db.Column(_db.Integer, primary_key=True)
+        addresses = _db.relationship(
+            'Address',
+            back_populates='account',
+        )
+
+    class Address(_db.Model):
+        __tablename__ = 'address'
+
+        id = _db.Column(_db.Integer, primary_key=True)
+
+        account_id = _db.Column(_db.Integer, _db.ForeignKey('account.id'))
+        account = _db.relationship('Account', back_populates='addresses')
+
+    # Create tables
+    _db.create_all()
+
+    @request.addfinalizer
+    def drop_tables():
+        _db.drop_all()
+
+    return Account, Address

tests/test_fixtures.py

@@ -294,3 +294,45 @@ def test_raw_connection_changes_dont_persist(person, db_engine, db_session):
 
     result = db_testdir.runpytest()
     result.assert_outcomes(passed=2)
+
+
+def test_commit_works_with_deleted_dependent(db_testdir):
+    '''
+    Make sure a commit still works with a dangling reference to a
+    deleted instance.
+
+    Exercise one way to trigger the issue reported in
+    https://github.com/jeancochrane/pytest-flask-sqlalchemy/issues/5
+    '''
+    db_testdir.makepyfile("""
+        def test_delete_message(account_address, db_session):
+            account, address = account_address
+
+            # Create a new object instance using the ORM
+            account_inst = account(id=1)
+
+            db_session.add(account_inst)
+
+            # Create a dependent object instance using the ORM
+
+            address_inst = address(id=101, account_id=1)
+
+            db_session.add(address_inst)
+            db_session.commit()
+
+            # Access the address through an ORM attribute
+            assert account_inst.addresses
+
+            # Delete the address out from under account_inst and the ORM
+            db_session.delete(address_inst)
+
+            # shouldn't see an exception like
+            #   sqlalchemy.exc.InvalidRequestError:
+            #   Instance XXX has been deleted.
+            #   Use the make_transient() function to send
+            #   this object back to the transient state.
+            db_session.commit()
+    """)
+
+    result = db_testdir.runpytest()
+    result.assert_outcomes(passed=1)