def test_no_instance_level_collections(self):
        @event.listens_for(self.Target, "event_one")
        def listen_one(x, y):
            pass

        t1 = self.Target()
        t2 = self.Target()
        t1.dispatch.event_one(5, 6)
        t2.dispatch.event_one(5, 6)
        is_(
            self.Target.dispatch._empty_listener_reg[self.Target]["event_one"],
            t1.dispatch.event_one,
        )

        @event.listens_for(t1, "event_one")
        def listen_two(x, y):
            pass

        is_not_(
            self.Target.dispatch._empty_listener_reg[self.Target]["event_one"],
            t1.dispatch.event_one,
        )
        is_(
            self.Target.dispatch._empty_listener_reg[self.Target]["event_one"],
            t2.dispatch.event_one,
        )

    def test_backref_events(self):
        User, Address = self._user_address_fixture(addresses_args={"backref": "user"})

        u1 = User()
        a1 = Address()
        u1.addresses.append(a1)
        is_(a1.user, u1)

    def test_select(self):
        t = Table("t", MetaData(), Column("x", Integer))
        s = t.select()

        is_(inspect(s), s)
        assert s.is_selectable
        is_(s.selectable, s)

    def test_inplace_add(self):
        User = self.classes.User
        session = Session()

        def l1():
            return session.query(User)

        def l2(q):
            return q.filter(User.name == bindparam('name'))

        q1 = self.bakery(l1)
        self._assert_cache_key(
            q1._cache_key,
            [l1]
        )
        eq_(q1.steps, [l1])

        q2 = q1.add_criteria(l2)
        is_(q2, q1)

        self._assert_cache_key(
            q1._cache_key,
            [l1, l2]
        )
        eq_(q1.steps, [l1, l2])

    def _run_test(self, specs, attributes):
        columns = [Column("c%i" % (i + 1), t[0]) for i, t in enumerate(specs)]

        # Early 5.0 releases seem to report more "general" for columns
        # in a view, e.g. char -> varchar, tinyblob -> mediumblob
        use_views = testing.db.dialect.server_version_info > (5, 0, 10)

        m = self.metadata
        Table("mysql_types", m, *columns)

        if use_views:
            event.listen(m, "after_create", DDL("CREATE OR REPLACE VIEW mysql_types_v " "AS SELECT * from mysql_types"))
            event.listen(m, "before_drop", DDL("DROP VIEW IF EXISTS mysql_types_v"))
        m.create_all()

        m2 = MetaData(testing.db)
        tables = [Table("mysql_types", m2, autoload=True)]
        if use_views:
            tables.append(Table("mysql_types_v", m2, autoload=True))

        for table in tables:
            for i, (reflected_col, spec) in enumerate(zip(table.c, specs)):
                expected_spec = spec[1]
                reflected_type = reflected_col.type
                is_(type(reflected_type), type(expected_spec))

                for attr in attributes:
                    eq_(
                        getattr(reflected_type, attr),
                        getattr(expected_spec, attr),
                        "Column %s: Attribute %s value of %s does not "
                        "match %s for type %s"
                        % ("c%i" % (i + 1), attr, getattr(reflected_type, attr), getattr(expected_spec, attr), spec[0]),
                    )

    def _test_round_trip(self, fixture, warnings=False):
        from sqlalchemy import inspect
        conn = testing.db.connect()
        for from_, to_ in self._fixture_as_string(fixture):
            inspector = inspect(conn)
            conn.execute("CREATE TABLE foo (data %s)" % from_)
            try:
                if warnings:
                    def go():
                        return inspector.get_columns("foo")[0]
                    col_info = testing.assert_warnings(go,
                                    ["Could not instantiate"], regex=True)
                else:
                    col_info = inspector.get_columns("foo")[0]
                expected_type = type(to_)
                is_(type(col_info['type']), expected_type)

                # test args
                for attr in ("scale", "precision", "length"):
                    if getattr(to_, attr, None) is not None:
                        eq_(
                            getattr(col_info['type'], attr),
                            getattr(to_, attr, None)
                        )
            finally:
                conn.execute("DROP TABLE foo")

    def test_sets(self):
# start Py2K
#        import sets
# end Py2K

        class SetLike(object):
            def add(self):
                pass

        class ForcedSet(list):
            __emulates__ = set

        for type_ in (set,
# start Py2K
#                      sets.Set,
# end Py2K
                      SetLike,
                      ForcedSet):
            eq_(util.duck_type_collection(type_), set)
            instance = type_()
            eq_(util.duck_type_collection(instance), set)

        for type_ in (frozenset,
# start Py2K
#                      sets.ImmutableSet
# end Py2K
                      ):
            is_(util.duck_type_collection(type_), None)
            instance = type_()
            is_(util.duck_type_collection(instance), None)

    def test_synonym_filter(self):
        User = self.classes.User
        syn = inspect(User).synonyms

        eq_(list(syn.keys()), ["name_syn"])
        is_(syn.name_syn, User.name_syn.original_property)
        eq_(dict(syn), {"name_syn": User.name_syn.original_property})

    def test_insp_column_prop(self):
        User = self.classes.User
        prop = inspect(User.name)
        is_(prop, User.name)

        is_(prop.parent, class_mapper(User))
        assert not hasattr(prop, "mapper")

    def test_password_custom_obj(self):
        class SecurePassword(str):
            def __init__(self, value):
                self.value = value

            def __str__(self):
                return self.value

        sp = SecurePassword("secured_password")
        u = url.URL("dbtype", username="x", password=sp, host="localhost")

        eq_(u.password, "secured_password")
        eq_(str(u), "dbtype://x:[email protected]")

        # test in-place modification
        sp.value = "new_secured_password"
        eq_(u.password, "new_secured_password")
        eq_(str(u), "dbtype://x:[email protected]")

        u.password = "hi"

        eq_(u.password, "hi")
        eq_(str(u), "dbtype://x:[email protected]")

        u.password = None

        is_(u.password, None)
        eq_(str(u), "dbtype://[email protected]")

    def test_invocation_systemwide_loaders(self):
        baked.bake_lazy_loaders()
        try:
            User, Address = self._o2m_fixture()

            sess = Session()
            q = sess.query(User).options(lazyload(User.addresses))
            with mock.patch.object(BakedLazyLoader, "_emit_lazyload") as el:
                u1 = q.first()
                u1.addresses
                # invoked
                is_(
                    el.mock_calls[0][1][1],
                    u1._sa_instance_state
                )
        finally:
            baked.unbake_lazy_loaders()

        clear_mappers()
        User, Address = self._o2m_fixture()
        sess = Session()
        q = sess.query(User).options(lazyload(User.addresses))

        with mock.patch.object(BakedLazyLoader, "_emit_lazyload") as el:
            u1 = q.first()
            u1.addresses
            # not invoked
            eq_(el.mock_calls, [])

    def test_any_literal(self):
        stuff = self.tables.stuff
        stmt = select([4 == any_(select([stuff.c.value]))])

        is_(
            testing.db.execute(stmt).scalar(), True
        )

    def test_post_update_m2o_detect_none(self):
        person, ball, Ball, Person = (
            self.tables.person,
            self.tables.ball,
            self.classes.Ball,
            self.classes.Person)

        mapper(Ball, ball, properties={
            'person': relationship(
                Person, post_update=True,
                primaryjoin=person.c.id == ball.c.person_id)
        })
        mapper(Person, person)

        sess = create_session(autocommit=False, expire_on_commit=True)
        sess.add(Ball(person=Person()))
        sess.commit()
        b1 = sess.query(Ball).first()

        # needs to be unloaded
        assert 'person' not in b1.__dict__
        b1.person = None

        self.assert_sql_execution(
            testing.db,
            sess.flush,
            CompiledSQL(
                "UPDATE ball SET person_id=:person_id "
                "WHERE ball.id = :ball_id",
                lambda ctx: {'person_id': None, 'ball_id': b1.id})
        )

        is_(b1.person, None)

    def test_savepoint_lost_still_runs(self):
        User = self.classes.User
        s = self.session(bind=self.bind)
        trans = s.begin_nested()
        s.connection()
        u1 = User(name='ed')
        s.add(u1)

        # kill off the transaction
        nested_trans = trans._connections[self.bind][1]
        nested_trans._do_commit()

        is_(s.transaction, trans)
        assert_raises(
            sa_exc.DBAPIError,
            s.rollback
        )

        assert u1 not in s.new

        is_(trans._state, _session.CLOSED)
        is_not_(s.transaction, trans)
        is_(s.transaction._state, _session.ACTIVE)

        is_(s.transaction.nested, False)

        is_(s.transaction._parent, None)

        def insert_values(engine, table_, values):
            """
            Inserts a row into a table, returns the full list of values
            INSERTed including defaults that fired off on the DB side and
            detects rows that had defaults and post-fetches.
            """

            # verify implicit_returning is working
            if engine.dialect.implicit_returning:
                ins = table_.insert()
                comp = ins.compile(engine, column_keys=list(values))
                if not set(values).issuperset(
                        c.key for c in table_.primary_key):
                    is_(bool(comp.returning), True)

            result = engine.execute(table_.insert(), **values)
            ret = values.copy()

            for col, id in zip(
                    table_.primary_key, result.inserted_primary_key):
                ret[col.key] = id

            if result.lastrow_has_defaults():
                criterion = and_(
                    *[
                        col == id for col, id in
                        zip(table_.primary_key, result.inserted_primary_key)])
                row = engine.execute(table_.select(criterion)).first()
                for c in table_.c:
                    ret[c.key] = row[c]
            return ret

    def test_column_in_mapper_args_used_multiple_times(self):

        class MyMixin(object):

            version_id = Column(Integer)
            __mapper_args__ = {'version_id_col': version_id}

        class ModelOne(Base, MyMixin):

            __tablename__ = 'm1'
            id = Column(Integer, primary_key=True)

        class ModelTwo(Base, MyMixin):

            __tablename__ = 'm2'
            id = Column(Integer, primary_key=True)

        is_(
            ModelOne.__mapper__.version_id_col,
            ModelOne.__table__.c.version_id
        )
        is_(
            ModelTwo.__mapper__.version_id_col,
            ModelTwo.__table__.c.version_id
        )

 def _run_test(self, specs, attributes):
     columns = [Column("c%i" % (i + 1), t[0]) for i, t in enumerate(specs)]
     m = self.metadata
     Table("oracle_types", m, *columns)
     m.create_all()
     m2 = MetaData(testing.db)
     table = Table("oracle_types", m2, autoload=True)
     for i, (reflected_col, spec) in enumerate(zip(table.c, specs)):
         expected_spec = spec[1]
         reflected_type = reflected_col.type
         is_(type(reflected_type), type(expected_spec))
         for attr in attributes:
             eq_(
                 getattr(reflected_type, attr),
                 getattr(expected_spec, attr),
                 "Column %s: Attribute %s value of %s does not "
                 "match %s for type %s"
                 % (
                     "c%i" % (i + 1),
                     attr,
                     getattr(reflected_type, attr),
                     getattr(expected_spec, attr),
                     spec[0],
                 ),
             )

    def test_column_error_printing(self):
        result = testing.db.execute(select([1]))
        row = result.first()

        class unprintable(object):

            def __str__(self):
                raise ValueError("nope")

        msg = r"Could not locate column in row for column '%s'"

        for accessor, repl in [
            ("x", "x"),
            (Column("q", Integer), "q"),
            (Column("q", Integer) + 12, r"q \+ :q_1"),
            (unprintable(), "unprintable element.*"),
        ]:
            assert_raises_message(
                exc.NoSuchColumnError,
                msg % repl,
                result._getter, accessor
            )

            is_(result._getter(accessor, False), None)

            assert_raises_message(
                exc.NoSuchColumnError,
                msg % repl,
                lambda: row[accessor]
            )

 def test_mapper_selectable(self):
     User = self.classes.User
     user_table = self.tables.users
     insp = inspect(User)
     is_(insp.selectable, user_table)
     assert not insp.is_selectable
     assert not insp.is_aliased_class

    def test_info_from_hybrid(self):
        A = self._fixture()
        A._value.info['foo'] = 'bar'
        A.value.info['bar'] = 'hoho'

        insp = inspect(A)
        is_(insp.all_orm_descriptors['value'].info, A.value.info)