sqlalchemy test cases for bug in listen() on Engine events fired on the Connection

test_listen.py

import sqlalchemy
from sqlalchemy import create_engine, Column, Integer
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

def setup():
    # set up a simple in memory sqlite db with two table
    engine = create_engine('sqlite:///:memory:')
    Base = declarative_base(bind=engine)
    class A(Base):
        __tablename__ = "a"
        id = Column("id", Integer, primary_key = True)

    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)

    return engine, A, Session

def test_session_before_listen():
    engine, A, Session = setup()

    got_event = []
    def before_execute_cb(*args):
        got_event.append(True)

    sess = Session()
    sqlalchemy.event.listen(engine, "before_execute", before_execute_cb)
    sess.query(A).all()

    # OK
    assert len(got_event) == 1

def test_query_before_listen():
    engine, A, Session = setup()

    got_event = []
    def before_execute_cb(*args):
        got_event.append(True)

    sess = Session()
    sess.query(A).all()
    sqlalchemy.event.listen(engine, "before_execute", before_execute_cb)
    sess2 = Session()
    sess2.query(A).all()

    # OK
    assert len(got_event) == 1

def test_query_before_listen_same_session():
    engine, A, Session = setup()

    got_event = []
    def before_execute_cb(*args):
        got_event.append(True)

    sess = Session()
    sess.query(A).all()
    sqlalchemy.event.listen(engine, "before_execute", before_execute_cb)
    sess.query(A).all() # <--- USE THE SAME SESSION AS BEFORE

    # FAIL
    assert len(got_event) == 1

def test_query_with_pre_registered_dummy_listener():
    engine, A, Session = setup()

    got_event = []
    def before_execute_cb(*args):
        got_event.append(True)

    def foo(*args):
        pass

    sqlalchemy.event.listen(engine, "before_execute", foo) # <--- Register a dummy listener to force sqlalchemy to join the Connection's dispatch descriptor with the Engine's
    sess = Session()
    sess.query(A).all()
    sqlalchemy.event.listen(engine, "before_execute", before_execute_cb)
    sess.query(A).all() # <--- USE THE SAME SESSION AS BEFORE
    
    # OK
    assert len(got_event) == 1