def test_parent_instance_child_class_apply_after(self):
        l1 = Mock()
        l2 = Mock()

        event.listen(self.TargetElement, "event_one", l2)

        factory = self.TargetFactory()
        element = factory.create()

        element.run_event(1)

        event.listen(factory, "event_one", l1)

        element.run_event(2)
        element.run_event(3)

        # c1 gets no events due to _JoinedListener
        # fixing the "parent" at construction time.
        # this can be changed to be "live" at the cost
        # of performance.
        eq_(
            l1.mock_calls, []
        )
        eq_(
            l2.mock_calls,
            [call(element, 1), call(element, 2), call(element, 3)]
        )

 def register_signals(self, app):
     """Register the signals."""
     before_record_index.connect(inject_provisional_community)
     if app.config['COMMUNITIES_OAI_ENABLED']:
         listen(Community, 'after_insert', create_oaipmh_set)
         listen(Community, 'after_delete', destroy_oaipmh_set)
     inclusion_request_created.connect(new_request)

 def init_sqlalchemy(self, scheme, connection):
     try:
         import sqlalchemy
         from sqlalchemy import create_engine, MetaData
         from sqlalchemy.orm import scoped_session, sessionmaker
         from torweb.db import CacheQuery
         import _mysql_exceptions
         from sqlalchemy import event
         from sqlalchemy.exc import DisconnectionError
         def my_on_checkout(dbapi_conn, connection_rec, connection_proxy):
             try:
                 dbapi_conn.cursor().execute('select now()')
             except _mysql_exceptions.OperationalError:
                 raise DisconnectionError
         
         engine = create_engine(
             connection,
             convert_unicode=True,
             encoding="utf-8",
             pool_recycle=3600*7,
             #echo_pool=True,
             echo=False,
         )
         event.listen(engine, 'checkout', my_on_checkout)
         metadata = MetaData(bind=engine)
         session = scoped_session(sessionmaker(bind=engine, query_cls=CacheQuery))
         sqlalchemy_sessions = [session]
         DB_Session = sessionmaker(bind=engine)
         return {"metadata":metadata, "session":session, "sqlalchemy_sessions":sqlalchemy_sessions}
         #setattr(self.app, 'metadata', metadata)
         #setattr(self.app, scheme.get('sqlalchemy', 'session'), session)
         #setattr(self.app, 'sqlalchemy_sessions', sqlalchemy_sessions)
     except Exception as e:
         print e

    def test_all_events(self):
        canary = []
        def before_attach(obj, parent):
            canary.append("%s->%s" % (obj.__class__.__name__, parent.__class__.__name__))

        def after_attach(obj, parent):
            canary.append("%s->%s" % (obj.__class__.__name__, parent))

        event.listen(schema.SchemaItem, "before_parent_attach", before_attach)
        event.listen(schema.SchemaItem, "after_parent_attach", after_attach)

        m = MetaData()
        t1 = Table('t1', m, 
            Column('id', Integer, Sequence('foo_id'), primary_key=True),
            Column('bar', String, ForeignKey('t2.id'))
        )
        t2 = Table('t2', m,
            Column('id', Integer, primary_key=True),
        )

        eq_(
            canary,
            ['Sequence->Column', 'Sequence->id', 'ForeignKey->Column',
             'ForeignKey->bar', 'Table->MetaData',
             'PrimaryKeyConstraint->Table', 'PrimaryKeyConstraint->t1',
             'Column->Table', 'Column->t1', 'Column->Table',
             'Column->t1', 'ForeignKeyConstraint->Table',
             'ForeignKeyConstraint->t1', 'Table->MetaData(bind=None)',
             'Table->MetaData', 'PrimaryKeyConstraint->Table',
             'PrimaryKeyConstraint->t2', 'Column->Table', 'Column->t2',
             'Table->MetaData(bind=None)']
        )

    def listen(self, cls, prp, handler):
        if (cls, prp) not in self.handlers:
            self.handlers[(cls, prp)] = list()
            event.listen(class_mapper(cls)._props[prp],
                         'set', self.make_callback(cls, prp))

        self.handlers[(cls, prp)].append(handler)

    def test_instance(self):
        Target = self._fixture()

        class Foo(object):
            def __init__(self):
                self.mock = Mock()

            def evt(self, arg):
                self.mock(arg)

        f1 = Foo()
        f2 = Foo()

        event.listen(Target, "event_one", f1.evt)
        event.listen(Target, "event_one", f2.evt)

        t1 = Target()
        t1.dispatch.event_one("x")

        event.remove(Target, "event_one", f1.evt)

        t1.dispatch.event_one("y")

        eq_(f1.mock.mock_calls, [call("x")])
        eq_(f2.mock.mock_calls, [call("x"), call("y")])

    def test_listener_collection_removed_cleanup(self):
        from sqlalchemy.event import registry

        Target = self._fixture()

        m1 = Mock()

        t1 = Target()

        event.listen(t1, "event_one", m1)

        key = (id(t1), "event_one", id(m1))

        assert key in registry._key_to_collection
        collection_ref = list(registry._key_to_collection[key])[0]
        assert collection_ref in registry._collection_to_key

        t1.dispatch.event_one("t1")

        del t1

        gc_collect()

        assert key not in registry._key_to_collection
        assert collection_ref not in registry._collection_to_key

def file_column(column):
    def set_event_listner(target, value, oldvalue, initiator):
        if oldvalue:
            full_path = FileProcess.fullPath(oldvalue)
            if (os.path.isfile(full_path)):
                os.unlink(full_path)
    alchemy_event.listen(column, 'set', set_event_listner)

    def __declare_last__(cls) -> None:
        """ Executed by SQLAlchemy as part of mapper configuration

        When the data changes, we want to reset the cursor position of target
        instance, to allow proper streaming of data.
        """
        event.listen(DBStoredFile.data, "set", handle_change_data)

def initialize_sql_test(engine):
    DBSession = scoped_session(sessionmaker(extension=ZopeTransactionExtension()))
    DBSession.configure(bind=engine)
    Base.metadata.bind = engine
    Base.metadata.create_all(engine)
    listen(sqlalchemy.orm.mapper, 'before_insert', make_uuid)
    return [DBSession, Base]

    def define_temp_tables(cls, metadata):
        # cheat a bit, we should fix this with some dialect-level
        # temp table fixture
        if testing.against("oracle"):
            kw = {
                'prefixes': ["GLOBAL TEMPORARY"],
                'oracle_on_commit': 'PRESERVE ROWS'
            }
        else:
            kw = {
                'prefixes': ["TEMPORARY"],
            }

        user_tmp = Table(
            "user_tmp", metadata,
            Column("id", sa.INT, primary_key=True),
            Column('name', sa.VARCHAR(50)),
            Column('foo', sa.INT),
            sa.UniqueConstraint('name', name='user_tmp_uq'),
            sa.Index("user_tmp_ix", "foo"),
            **kw
        )
        if testing.requires.view_reflection.enabled and \
                testing.requires.temporary_views.enabled:
            event.listen(
                user_tmp, "after_create",
                DDL("create temporary view user_tmp_v as "
                    "select * from user_tmp")
            )
            event.listen(
                user_tmp, "before_drop",
                DDL("drop view user_tmp_v")
            )

    def test_transactional_advanced(self):
        canary = []
        def tracker(name):
            def go(*args, **kw):
                canary.append(name)
            return go

        engine = engines.testing_engine()
        for name in ['begin', 'savepoint', 
                    'rollback_savepoint', 'release_savepoint',
                    'rollback', 'begin_twophase', 
                       'prepare_twophase', 'commit_twophase']:
            event.listen(engine, '%s' % name, tracker(name))

        conn = engine.connect()

        trans = conn.begin()
        trans2 = conn.begin_nested()
        conn.execute(select([1]))
        trans2.rollback()
        trans2 = conn.begin_nested()
        conn.execute(select([1]))
        trans2.commit()
        trans.rollback()

        trans = conn.begin_twophase()
        conn.execute(select([1]))
        trans.prepare()
        trans.commit()

        eq_(canary, ['begin', 'savepoint', 
                    'rollback_savepoint', 'savepoint', 'release_savepoint',
                    'rollback', 'begin_twophase', 
                       'prepare_twophase', 'commit_twophase']
        )

    def test_transactional(self):
        canary = []
        def tracker(name):
            def go(conn, *args, **kw):
                canary.append(name)
            return go

        engine = engines.testing_engine()
        event.listen(engine, 'before_execute', tracker('execute'))
        event.listen(engine, 'before_cursor_execute', tracker('cursor_execute'))
        event.listen(engine, 'begin', tracker('begin'))
        event.listen(engine, 'commit', tracker('commit'))
        event.listen(engine, 'rollback', tracker('rollback'))

        conn = engine.connect()
        trans = conn.begin()
        conn.execute(select([1]))
        trans.rollback()
        trans = conn.begin()
        conn.execute(select([1]))
        trans.commit()

        eq_(canary, [
            'begin', 'execute', 'cursor_execute', 'rollback',
            'begin', 'execute', 'cursor_execute', 'commit',
            ])

    def test_retval_flag(self):
        canary = []
        def tracker(name):
            def go(conn, *args, **kw):
                canary.append(name)
            return go

        def execute(conn, clauseelement, multiparams, params):
            canary.append('execute')
            return clauseelement, multiparams, params

        def cursor_execute(conn, cursor, statement, 
                        parameters, context, executemany):
            canary.append('cursor_execute')
            return statement, parameters

        engine = engines.testing_engine()

        assert_raises(
            tsa.exc.ArgumentError,
            event.listen, engine, "begin", tracker("begin"), retval=True
        )

        event.listen(engine, "before_execute", execute, retval=True)
        event.listen(engine, "before_cursor_execute", cursor_execute, retval=True)
        engine.execute(select([1]))
        eq_(
            canary, ['execute', 'cursor_execute']
        )

 def register(self):
     event.listen(
         self.engine, 'before_cursor_execute', self.before_cursor_execute
     )
     event.listen(
         self.engine, 'after_cursor_execute', self.after_cursor_execute
     )

def includeme(config: Configurator) -> None:
    """ Pyramid includeme hook.

    :param config: app config
    :type config: :class:`pyramid.config.Configurator`
    """

    config.add_tween(
        "kotti.filedepot.TweenFactory", over=tweens.MAIN, under=tweens.INGRESS
    )
    config.add_request_method(uploaded_file_response, name="uploaded_file_response")
    config.add_request_method(uploaded_file_url, name="uploaded_file_url")

    from kotti.events import objectevent_listeners
    from kotti.events import ObjectInsert
    from kotti.events import ObjectUpdate

    from sqlalchemy.event import listen
    from sqlalchemy.engine import Engine

    listen(Engine, "engine_connect", adjust_for_engine)

    configure_filedepot(config.get_settings())

    # Update file metadata on change of blob data
    objectevent_listeners[(ObjectInsert, DBStoredFile)].append(set_metadata)
    objectevent_listeners[(ObjectUpdate, DBStoredFile)].append(set_metadata)

    # depot's _SQLAMutationTracker._session_committed is executed on
    # after_commit, that's too late for DBFileStorage to interact with the
    # session
    event.listen(DBSession, "before_commit", _SQLAMutationTracker._session_committed)

def handle_error(engine, listener):
    """Add a handle_error listener for the given :class:`.Engine`.

    This listener uses the SQLAlchemy
    :meth:`sqlalchemy.event.ConnectionEvents.handle_error`
    event, however augments the listener for pre-0.9.7 versions of SQLAlchemy
    in order to support safe re-raise of the exception.

    """
    if utils.sqla_100:
        event.listen(engine, "handle_error", listener)
        return

    assert isinstance(engine, Engine), \
        "engine argument must be an Engine instance, not a Connection"

    if not utils.sqla_097:
        _rework_handle_exception_for_events(engine)
        engine._oslo_handle_error_events.append(listener)

    _rework_connect_and_revalidate_for_events(engine)

    if utils.sqla_097:
        # ctx.engine added per
        # https://bitbucket.org/zzzeek/sqlalchemy/issue/3266/
        def wrap_listener(ctx):
            if isinstance(ctx, engine_base.ExceptionContextImpl):
                ctx.engine = ctx.connection.engine
            return listener(ctx)
        event.listen(engine, "handle_error", wrap_listener)

        def test_before_update_m2o(self):
            """Expect normal many to one attribute load behavior
            (should not get committed value)
            from within public 'before_update' event"""
            sess = self._mapper_setup()

            Address, User = self.classes.Address, self.classes.User

            def before_update(mapper, connection, target):
                # if get committed is used to find target.user, then
                # it will be still be u1 instead of u2
                assert target.user.id == target.user_id == u2.id
            from sqlalchemy import event
            event.listen(Address, 'before_update', before_update)

            a1 = Address(email_address='a1')
            u1 = User(name='u1', addresses=[a1])
            sess.add(u1)

            u2 = User(name='u2')
            sess.add(u2)
            sess.commit()

            sess.expunge_all()
            # lookup an address and move it to the other user
            a1 = sess.query(Address).get(a1.id)

            # move address to another user's fk
            assert a1.user_id == u1.id
            a1.user_id = u2.id

            sess.flush()

    def test_propagate(self):
        Target = self._fixture()

        m1 = Mock()

        t1 = Target()
        t2 = Target()

        event.listen(t1, "event_one", m1, propagate=True)
        event.listen(t1, "event_two", m1, propagate=False)

        t2.dispatch._update(t1.dispatch)

        t1.dispatch.event_one("t1e1x")
        t1.dispatch.event_two("t1e2x")
        t2.dispatch.event_one("t2e1x")
        t2.dispatch.event_two("t2e2x")

        event.remove(t1, "event_one", m1)
        event.remove(t1, "event_two", m1)

        t1.dispatch.event_one("t1e1y")
        t1.dispatch.event_two("t1e2y")
        t2.dispatch.event_one("t2e1y")
        t2.dispatch.event_two("t2e2y")

        eq_(m1.mock_calls,
                [call('t1e1x'), call('t1e2x'),
                call('t2e1x')])

    def test_accounting_commit_fails_delete(self):
        User = self.classes.User
        sess = create_session(autocommit=True)

        fail = False

        def fail_fn(*arg, **kw):
            if fail:
                raise Exception("commit fails")

        event.listen(sess, "after_flush_postexec", fail_fn)
        u1 = User(name='ed')
        sess.add(u1)
        sess.flush()

        sess.delete(u1)
        fail = True
        assert_raises(
            Exception,
            sess.flush
        )
        fail = False

        assert u1 in sess
        assert u1 not in sess.deleted
        sess.delete(u1)
        sess.flush()
        assert u1 not in sess
        eq_(
            sess.query(User.name).order_by(User.name).all(),
            []
        )