def test_warning_on_using_inactive_session_rollback_evt(self):
        users, User = self.tables.users, self.classes.User

        mapper(User, users)
        sess = Session()
        u1 = User(id=1, name='u1')
        sess.add(u1)
        sess.commit()

        u3 = User(name='u3')

        @event.listens_for(sess, "after_rollback")
        def evt(s):
            sess.add(u3)

        sess.add(User(id=1, name='u2'))

        def go():
            assert_raises(
                orm_exc.FlushError, sess.flush
            )

        assert_warnings(go,
                        ["Session's state has been changed on a "
                         "non-active transaction - this state "
                         "will be discarded."],
                        )
        assert u3 not in sess

    def test_columnadapter_anonymized(self):
        """test issue #3148

        Testing the anonymization applied from the ColumnAdapter.columns
        collection, typically as used in eager loading.

        """
        exprs = [
            table1.c.myid,
            table1.c.name.label('t1name'),
            func.foo("hoho").label('x')]

        ta = table1.alias()
        adapter = sql_util.ColumnAdapter(ta, anonymize_labels=True)

        s1 = select([adapter.columns[expr] for expr in exprs]).\
            apply_labels().order_by("myid", "t1name", "x")

        def go():
            # the labels here are anonymized, so label naming
            # can't catch these.
            self.assert_compile(
                s1,
                "SELECT mytable_1.myid AS mytable_1_myid, "
                "mytable_1.name AS name_1, foo(:foo_2) AS foo_1 "
                "FROM mytable AS mytable_1 ORDER BY mytable_1.myid, t1name, x"
            )

        assert_warnings(
            go,
            ["Can't resolve label reference 't1name'",
             "Can't resolve label reference 'x'"], regex=True)

    def test_index_reflection(self):
        """ Reflecting partial & expression-based indexes should warn
        """

        metadata = self.metadata

        t1 = Table(
            "party",
            metadata,
            Column("id", String(10), nullable=False),
            Column("name", String(20), index=True),
            Column("aname", String(20)),
        )
        metadata.create_all()
        testing.db.execute(
            """
          create index idx1 on party ((id || name))
        """
        )
        testing.db.execute(
            """
          create unique index idx2 on party (id) where name = 'test'
        """
        )
        testing.db.execute(
            """
            create index idx3 on party using btree
                (lower(name::text), lower(aname::text))
        """
        )

        def go():
            m2 = MetaData(testing.db)
            t2 = Table("party", m2, autoload=True)
            assert len(t2.indexes) == 2

            # Make sure indexes are in the order we expect them in

            tmp = [(idx.name, idx) for idx in t2.indexes]
            tmp.sort()
            r1, r2 = [idx[1] for idx in tmp]
            assert r1.name == "idx2"
            assert r1.unique is True
            assert r2.unique is False
            assert [t2.c.id] == r1.columns
            assert [t2.c.name] == r2.columns

        testing.assert_warnings(
            go,
            [
                "Skipped unsupported reflection of "
                "expression-based index idx1",
                "Predicate of partial index idx2 ignored during " "reflection",
                "Skipped unsupported reflection of "
                "expression-based index idx3",
            ],
        )

 def test_warning_on_using_inactive_session_delete(self):
     sess, u1 = self._inactive_flushed_session_fixture()
     sess.delete(u1)
     def go():
         sess.rollback()
     assert_warnings(go,
         ["Session's state has been changed on a "
         "non-active transaction - this state "
         "will be discarded."],
     )
     assert u1 in sess
     assert u1 not in sess.deleted

    def test_warning_on_using_inactive_session_new(self):
        User = self.classes.User

        sess, u1 = self._inactive_flushed_session_fixture()
        u2 = User(name='u2')
        sess.add(u2)
        def go():
            sess.rollback()
        assert_warnings(go,
            ["Session's state has been changed on a "
            "non-active transaction - this state "
            "will be discarded."],
        )
        assert u2 not in sess
        assert u1 in sess

    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_pyodbc_host_no_driver(self):
        dialect = pyodbc.dialect()
        u = url.make_url("mssql://username:[email protected]/database")

        def go():
            return dialect.create_connect_args(u)

        connection = assert_warnings(
            go,
            [
                "No driver name specified; this is expected by "
                "PyODBC when using DSN-less connections"
            ],
        )

        eq_(
            [
                [
                    "Server=hostspec;Database=database;UI"
                    "D=username;PWD=password"
                ],
                {},
            ],
            connection,
        )

 def _test_lookup_direct(self, fixture, warnings=False):
     dialect = sqlite.dialect()
     for from_, to_ in self._fixture_as_string(fixture):
         if warnings:
             def go():
                 return dialect._resolve_type_affinity(from_)
             final_type = testing.assert_warnings(go,
                             ["Could not instantiate"], regex=True)
         else:
             final_type = dialect._resolve_type_affinity(from_)
         expected_type = type(to_)
         is_(type(final_type), expected_type)