sqlalchemy invalidation test

test_invalidate.py

from __future__ import print_function

# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
#
# Use this file to build your own SSCCE
# SSCCE = Short, Self Contained, Correct (Compatible) Example
# see http://sscce.org/
#
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -


# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# Standard imports

import pdb

import sqlalchemy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import deferred, class_mapper
from sqlalchemy import Column, Integer
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import inspect
from sqlalchemy.orm.session import object_session

# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# You probably don't need to overwrite this
Base = declarative_base()


# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# Define some models that inherit from Base


class Foo(Base):
    __tablename__ = "foo"
    id = Column(Integer, primary_key=True)


# - - - -  - - - - - - - - - - - - - - - - - - - - - - - - - - -
# we'll use sqlite+memory to handle this
# if your issue is backend specific, then the engine will need to be different


sqlalchemy_url = "sqlite://"
# sqlalchemy_url = "postgresql://sa_test:sa_test@localhost/sqlalchemy_test"

engine = create_engine(sqlalchemy_url, echo=False)
Base.metadata.create_all(engine)


# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# you shouldn't change these 2 line

sessionFactory = sessionmaker(bind=engine)

# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# Start your example here:


# setup
for i in (1, 2):
    try:
        _s = sessionFactory()
        f = Foo()
        f.id = i
        _s.add(f)
        _s.commit()
    except Exception as exc:
        # duplicate errors likely in postgres
        pass


def test_original():
    """
    original testcase submitted in https://github.com/sqlalchemy/sqlalchemy/issues/5098
    
    Under Sqlite:

        1 conn= <sqlalchemy.engine.base.Connection object at 0x1022a0110>
        2 s= <sqlalchemy.orm.session.Session object at 0x1022a0210>
        3 r= <__main__.Foo object at 0x102704f90>
        4 post invalidate()
        Traceback (most recent call last):
            sqlalchemy.exc.StatementError: (sqlalchemy.exc.InvalidRequestError) Can't reconnect until invalid transaction is rolled back

    Under PostgreSQL
        1 conn= <sqlalchemy.engine.base.Connection object at 0x1026e5050>
        2 s= <sqlalchemy.orm.session.Session object at 0x102704250>
        3 r= <__main__.Foo object at 0x102704f90>
        4 post invalidate()
        Traceback (most recent call last):
            sqlalchemy.exc.StatementError: (sqlalchemy.exc.InvalidRequestError) Can't reconnect until invalid transaction is rolled back
    """

    print("test_original")
    conn = engine.connect()
    print(1, "conn=", conn)
    s = sessionFactory(bind=conn)
    print(2, "s=", s)
    r = s.query(Foo).filter_by(id=1).first()
    print(3, "r=", r)
    conn.invalidate()
    print(4, "post invalidate()")
    r = s.query(Foo).filter_by(id=1).first()
    print(5, "r=", r)


def test_explore():
    """
    """

    print("test_explore")

    conn = engine.connect()  # not necessary since Session will establish a connection
    print(1, "conn=", conn)

    s = sessionFactory(bind=conn)
    print(2, "s=", s)

    s_conn_1 = s.connection()  # the session may have it's own connection
    print(3, "s_conn_1=", s_conn_1)

    r1 = s.query(Foo).filter_by(id=1).first()
    print(4, "r1=", r1)

    r1_info = inspect(r1)
    print("r1_info.persistent", r1_info.persistent)
    obj_conn_1 = object_session(r1).connection()

    # make sure we have the same connection.
    # an early testcase didn't supply `bind=conn` to the Session, which broke this test
    assert (conn == s_conn_1) and (s_conn_1 == obj_conn_1)
    if False:
        print("object_session(r1).connection()", obj_conn_1)
        print(
            "object_session(r1).connection() == s.connection()", obj_conn_1 == s_conn_1
        )
        print("object_session(r1).connection() == engine.connect()", obj_conn_1 == conn)

    # invalidate and close things
    conn.invalidate()
    obj_conn_1.invalidate()
    print(5, "post invalidate()")

    if False:
        # under PostgreSQL, enabling this block will cause the query for `r2` to fail with the following error:
        # sqlalchemy.exc.StatementError: (sqlalchemy.exc.ResourceClosedError) This Connection is closed
        conn.close()
        obj_conn_1.close()
        print(6, "post close()")

    print(7, "s=", s)
    s_conn_2 = s.connection()  # the session may have it's own connection
    print(7, "s_conn_2=", s_conn_2)

    try:
        r2 = s.query(Foo).filter_by(id=2).first()
        raise ValueError("this should have failed!")
    except:
        # expected!
        raise


def test_concern():
    """
    """

    print("test_concern")

    s = sessionFactory()  # just use the engine for auto-connections
    print(2, "s=", s)

    s_conn_1 = s.connection()
    print(3, "s_conn_1=", s_conn_1)

    r1 = s.query(Foo).filter_by(id=1).first()
    print(4, "r1=", r1)

    r1_object_connection = object_session(r1).connection()
    print("5", "r1_object_connection=", r1_object_connection)

    assert r1_object_connection == s_conn_1

    # invalidate and close things
    s_conn_1.invalidate()
    if False:
        # if this doesn't run...
        # postgres/sqlite error on query below is:
        #    sqlalchemy.exc.StatementError: (sqlalchemy.exc.InvalidRequestError) Can't reconnect until invalid transaction is rolled back
        # if this does run...
        # postgres/sqlute error on query below is:
        #    sqlalchemy.exc.StatementError: (sqlalchemy.exc.ResourceClosedError) This Connection is closed
        s_conn_1.close()

    if False:
        # running either of these is what we want to do so the test will run
        s.close()
        s.invalidate()

    # now what happens if we keep going...
    print(7, "s=", s)
    s_conn_2 = s.connection()  # the session may have it's own connection

    print(8, "s_conn_2=", s_conn_2)

    # todo: the setup data gets cleared away in sqlite. not sure why. example holds in postgresql
    r2 = s.query(Foo).filter_by(id=2).first()
    print(9, "r2=", r2)

    r2_object_connection = object_session(r2).connection()
    print("10", "r2_object_connection=", r2_object_connection)

    assert r2_object_connection == s_conn_2
    assert r2_object_connection != r1_object_connection


test_concern()
exit()

if True:
    test_concern()
else:
    test_original()
    test_explore()