Refactor for cx_Oracle version 6oracle_numeric

Drops support for cx_Oracle prior to version 5.x, reworks
numeric and binary support.

Fixes: #4064

Change-Id: Ib9ae9aba430c15cd2a6eeb4e5e3fd8e97b5fe480

1 files changed, 0 insertions, 2316 deletions

diff --git a/test/dialect/test_oracle.py b/test/dialect/test_oracle.py
deleted file mode 100644
index 5ca95aea9..000000000
--- a/test/dialect/test_oracle.py
+++ /dev/null
@@ -1,2316 +0,0 @@
-# coding: utf-8
-
-
-from sqlalchemy.testing import eq_
-from sqlalchemy import *
-from sqlalchemy import types as sqltypes, exc, schema
-from sqlalchemy.sql import table, column
-from sqlalchemy.sql.elements import quoted_name
-from sqlalchemy.testing import (fixtures,
-                                AssertsExecutionResults,
-                                AssertsCompiledSQL)
-from sqlalchemy import testing
-from sqlalchemy.util import u, b
-from sqlalchemy import util
-from sqlalchemy.testing import assert_raises, assert_raises_message
-from sqlalchemy.testing.engines import testing_engine
-from sqlalchemy.dialects.oracle import cx_oracle, base as oracle
-from sqlalchemy.engine import default
-import decimal
-from sqlalchemy.engine import url
-from sqlalchemy.testing.schema import Table, Column
-import datetime
-import os
-from sqlalchemy import sql
-from sqlalchemy.testing.mock import Mock
-
-
-class DialectTest(fixtures.TestBase):
-    def test_cx_oracle_version_parse(self):
-        dialect = cx_oracle.OracleDialect_cx_oracle()
-
-        eq_(
-            dialect._parse_cx_oracle_ver("5.2"),
-            (5, 2)
-        )
-
-        eq_(
-            dialect._parse_cx_oracle_ver("5.0.1"),
-            (5, 0, 1)
-        )
-
-        eq_(
-            dialect._parse_cx_oracle_ver("6.0b1"),
-            (6, 0)
-        )
-
-    def test_twophase_arg(self):
-
-        mock_dbapi = Mock(version="5.0.3")
-        dialect = cx_oracle.OracleDialect_cx_oracle(dbapi=mock_dbapi)
-        args = dialect.create_connect_args(
-            url.make_url("oracle+cx_oracle://a:b@host/db"))
-
-        eq_(args[1]['twophase'], True)
-
-        mock_dbapi = Mock(version="5.0.3")
-        dialect = cx_oracle.OracleDialect_cx_oracle(
-            dbapi=mock_dbapi, allow_twophase=False)
-        args = dialect.create_connect_args(
-            url.make_url("oracle+cx_oracle://a:b@host/db"))
-
-        eq_(args[1]['twophase'], False)
-
-        mock_dbapi = Mock(version="6.0b1")
-        dialect = cx_oracle.OracleDialect_cx_oracle(dbapi=mock_dbapi)
-        args = dialect.create_connect_args(
-            url.make_url("oracle+cx_oracle://a:b@host/db"))
-
-        assert 'twophase' not in args[1]
-
-
-class OutParamTest(fixtures.TestBase, AssertsExecutionResults):
-    __only_on__ = 'oracle+cx_oracle'
-    __backend__ = True
-
-    @classmethod
-    def setup_class(cls):
-        testing.db.execute("""
-        create or replace procedure foo(x_in IN number, x_out OUT number,
-        y_out OUT number, z_out OUT varchar) IS
-        retval number;
-        begin
-            retval := 6;
-            x_out := 10;
-            y_out := x_in * 15;
-            z_out := NULL;
-        end;
-        """)
-
-    def test_out_params(self):
-        result = testing.db.execute(text('begin foo(:x_in, :x_out, :y_out, '
-                                         ':z_out); end;',
-                                    bindparams=[bindparam('x_in', Float),
-                                                outparam('x_out', Integer),
-                                                outparam('y_out', Float),
-                                                outparam('z_out', String)]),
-                                    x_in=5)
-        eq_(result.out_parameters,
-            {'x_out': 10, 'y_out': 75, 'z_out': None})
-        assert isinstance(result.out_parameters['x_out'], int)
-
-    @classmethod
-    def teardown_class(cls):
-        testing.db.execute("DROP PROCEDURE foo")
-
-
-class CXOracleArgsTest(fixtures.TestBase):
-    __only_on__ = 'oracle+cx_oracle'
-    __backend__ = True
-
-    def test_autosetinputsizes(self):
-        dialect = cx_oracle.dialect()
-        assert dialect.auto_setinputsizes
-
-        dialect = cx_oracle.dialect(auto_setinputsizes=False)
-        assert not dialect.auto_setinputsizes
-
-    def test_exclude_inputsizes_none(self):
-        dialect = cx_oracle.dialect(exclude_setinputsizes=None)
-        eq_(dialect.exclude_setinputsizes, set())
-
-    def test_exclude_inputsizes_custom(self):
-        import cx_Oracle
-        dialect = cx_oracle.dialect(dbapi=cx_Oracle,
-                                    exclude_setinputsizes=('NCLOB',))
-        eq_(dialect.exclude_setinputsizes, set([cx_Oracle.NCLOB]))
-
-
-class QuotedBindRoundTripTest(fixtures.TestBase):
-
-    __only_on__ = 'oracle'
-    __backend__ = True
-
-    @testing.provide_metadata
-    def test_table_round_trip(self):
-        oracle.RESERVED_WORDS.remove('UNION')
-
-        metadata = self.metadata
-        table = Table("t1", metadata,
-                      Column("option", Integer),
-                      Column("plain", Integer, quote=True),
-                      # test that quote works for a reserved word
-                      # that the dialect isn't aware of when quote
-                      # is set
-                      Column("union", Integer, quote=True))
-        metadata.create_all()
-
-        table.insert().execute(
-            {"option": 1, "plain": 1, "union": 1}
-        )
-        eq_(
-            testing.db.execute(table.select()).first(),
-            (1, 1, 1)
-        )
-        table.update().values(option=2, plain=2, union=2).execute()
-        eq_(
-            testing.db.execute(table.select()).first(),
-            (2, 2, 2)
-        )
-
-    def test_numeric_bind_round_trip(self):
-        eq_(
-            testing.db.scalar(
-                select([
-                    literal_column("2", type_=Integer()) +
-                    bindparam("2_1", value=2)])
-            ),
-            4
-        )
-
-    @testing.provide_metadata
-    def test_numeric_bind_in_crud(self):
-        t = Table(
-            "asfd", self.metadata,
-            Column("100K", Integer)
-        )
-        t.create()
-
-        testing.db.execute(t.insert(), {"100K": 10})
-        eq_(
-            testing.db.scalar(t.select()), 10
-        )
-
-
-class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
-    __dialect__ = "oracle"  # oracle.dialect()
-
-    def test_true_false(self):
-        self.assert_compile(
-            sql.false(), "0"
-        )
-        self.assert_compile(
-            sql.true(),
-            "1"
-        )
-
-    def test_owner(self):
-        meta = MetaData()
-        parent = Table('parent', meta, Column('id', Integer,
-                       primary_key=True), Column('name', String(50)),
-                       schema='ed')
-        child = Table('child', meta, Column('id', Integer,
-                      primary_key=True), Column('parent_id', Integer,
-                      ForeignKey('ed.parent.id')), schema='ed')
-        self.assert_compile(parent.join(child),
-                            'ed.parent JOIN ed.child ON ed.parent.id = '
-                            'ed.child.parent_id')
-
-    def test_subquery(self):
-        t = table('sometable', column('col1'), column('col2'))
-        s = select([t])
-        s = select([s.c.col1, s.c.col2])
-
-        self.assert_compile(s, "SELECT col1, col2 FROM (SELECT "
-                               "sometable.col1 AS col1, sometable.col2 "
-                               "AS col2 FROM sometable)")
-
-    def test_bindparam_quote(self):
-        """test that bound parameters take on quoting for reserved words,
-        column names quote flag enabled."""
-        # note: this is only in cx_oracle at the moment.  not sure
-        # what other hypothetical oracle dialects might need
-
-        self.assert_compile(
-            bindparam("option"), ':"option"'
-        )
-        self.assert_compile(
-            bindparam("plain"), ':plain'
-        )
-        t = Table("s", MetaData(), Column('plain', Integer, quote=True))
-        self.assert_compile(
-            t.insert().values(plain=5),
-            'INSERT INTO s ("plain") VALUES (:"plain")'
-        )
-        self.assert_compile(
-            t.update().values(plain=5), 'UPDATE s SET "plain"=:"plain"'
-        )
-
-    def test_cte(self):
-        part = table(
-            'part',
-            column('part'),
-            column('sub_part'),
-            column('quantity')
-        )
-
-        included_parts = select([
-            part.c.sub_part, part.c.part, part.c.quantity
-        ]).where(part.c.part == "p1").\
-            cte(name="included_parts", recursive=True).\
-            suffix_with(
-                "search depth first by part set ord1",
-                "cycle part set y_cycle to 1 default 0", dialect='oracle')
-
-        incl_alias = included_parts.alias("pr1")
-        parts_alias = part.alias("p")
-        included_parts = included_parts.union_all(
-            select([
-                parts_alias.c.sub_part,
-                parts_alias.c.part, parts_alias.c.quantity
-            ]).where(parts_alias.c.part == incl_alias.c.sub_part)
-        )
-
-        q = select([
-            included_parts.c.sub_part,
-            func.sum(included_parts.c.quantity).label('total_quantity')]).\
-            group_by(included_parts.c.sub_part)
-
-        self.assert_compile(
-            q,
-            "WITH included_parts(sub_part, part, quantity) AS "
-            "(SELECT part.sub_part AS sub_part, part.part AS part, "
-            "part.quantity AS quantity FROM part WHERE part.part = :part_1 "
-            "UNION ALL SELECT p.sub_part AS sub_part, p.part AS part, "
-            "p.quantity AS quantity FROM part p, included_parts pr1 "
-            "WHERE p.part = pr1.sub_part) "
-            "search depth first by part set ord1 cycle part set "
-            "y_cycle to 1 default 0  "
-            "SELECT included_parts.sub_part, sum(included_parts.quantity) "
-            "AS total_quantity FROM included_parts "
-            "GROUP BY included_parts.sub_part"
-        )
-
-    def test_limit(self):
-        t = table('sometable', column('col1'), column('col2'))
-        s = select([t])
-        c = s.compile(dialect=oracle.OracleDialect())
-        assert t.c.col1 in set(c._create_result_map()['col1'][1])
-        s = select([t]).limit(10).offset(20)
-        self.assert_compile(s,
-                            'SELECT col1, col2 FROM (SELECT col1, '
-                            'col2, ROWNUM AS ora_rn FROM (SELECT '
-                            'sometable.col1 AS col1, sometable.col2 AS '
-                            'col2 FROM sometable) WHERE ROWNUM <= '
-                            ':param_1 + :param_2) WHERE ora_rn > :param_2',
-                            checkparams={'param_1': 10, 'param_2': 20})
-
-        c = s.compile(dialect=oracle.OracleDialect())
-        eq_(len(c._result_columns), 2)
-        assert t.c.col1 in set(c._create_result_map()['col1'][1])
-
-        s2 = select([s.c.col1, s.c.col2])
-        self.assert_compile(s2,
-                            'SELECT col1, col2 FROM (SELECT col1, col2 '
-                            'FROM (SELECT col1, col2, ROWNUM AS ora_rn '
-                            'FROM (SELECT sometable.col1 AS col1, '
-                            'sometable.col2 AS col2 FROM sometable) '
-                            'WHERE ROWNUM <= :param_1 + :param_2) '
-                            'WHERE ora_rn > :param_2)',
-                            checkparams={'param_1': 10, 'param_2': 20})
-
-        self.assert_compile(s2,
-                            'SELECT col1, col2 FROM (SELECT col1, col2 '
-                            'FROM (SELECT col1, col2, ROWNUM AS ora_rn '
-                            'FROM (SELECT sometable.col1 AS col1, '
-                            'sometable.col2 AS col2 FROM sometable) '
-                            'WHERE ROWNUM <= :param_1 + :param_2) '
-                            'WHERE ora_rn > :param_2)')
-        c = s2.compile(dialect=oracle.OracleDialect())
-        eq_(len(c._result_columns), 2)
-        assert s.c.col1 in set(c._create_result_map()['col1'][1])
-
-        s = select([t]).limit(10).offset(20).order_by(t.c.col2)
-        self.assert_compile(s,
-                            'SELECT col1, col2 FROM (SELECT col1, '
-                            'col2, ROWNUM AS ora_rn FROM (SELECT '
-                            'sometable.col1 AS col1, sometable.col2 AS '
-                            'col2 FROM sometable ORDER BY '
-                            'sometable.col2) WHERE ROWNUM <= '
-                            ':param_1 + :param_2) WHERE ora_rn > :param_2',
-                            checkparams={'param_1': 10, 'param_2': 20}
-                            )
-        c = s.compile(dialect=oracle.OracleDialect())
-        eq_(len(c._result_columns), 2)
-        assert t.c.col1 in set(c._create_result_map()['col1'][1])
-
-        s = select([t], for_update=True).limit(10).order_by(t.c.col2)
-        self.assert_compile(s,
-                            'SELECT col1, col2 FROM (SELECT '
-                            'sometable.col1 AS col1, sometable.col2 AS '
-                            'col2 FROM sometable ORDER BY '
-                            'sometable.col2) WHERE ROWNUM <= :param_1 '
-                            'FOR UPDATE')
-
-        s = select([t],
-                   for_update=True).limit(10).offset(20).order_by(t.c.col2)
-        self.assert_compile(s,
-                            'SELECT col1, col2 FROM (SELECT col1, '
-                            'col2, ROWNUM AS ora_rn FROM (SELECT '
-                            'sometable.col1 AS col1, sometable.col2 AS '
-                            'col2 FROM sometable ORDER BY '
-                            'sometable.col2) WHERE ROWNUM <= '
-                            ':param_1 + :param_2) WHERE ora_rn > :param_2 FOR '
-                            'UPDATE')
-
-    def test_for_update(self):
-        table1 = table('mytable',
-                       column('myid'), column('name'), column('description'))
-
-        self.assert_compile(
-            table1.select(table1.c.myid == 7).with_for_update(),
-            "SELECT mytable.myid, mytable.name, mytable.description "
-            "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE")
-
-        self.assert_compile(
-            table1
-            .select(table1.c.myid == 7)
-            .with_for_update(of=table1.c.myid),
-            "SELECT mytable.myid, mytable.name, mytable.description "
-            "FROM mytable WHERE mytable.myid = :myid_1 "
-            "FOR UPDATE OF mytable.myid")
-
-        self.assert_compile(
-            table1.select(table1.c.myid == 7).with_for_update(nowait=True),
-            "SELECT mytable.myid, mytable.name, mytable.description "
-            "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE NOWAIT")
-
-        self.assert_compile(
-            table1
-            .select(table1.c.myid == 7)
-            .with_for_update(nowait=True, of=table1.c.myid),
-            "SELECT mytable.myid, mytable.name, mytable.description "
-            "FROM mytable WHERE mytable.myid = :myid_1 "
-            "FOR UPDATE OF mytable.myid NOWAIT")
-
-        self.assert_compile(
-            table1
-            .select(table1.c.myid == 7)
-            .with_for_update(nowait=True, of=[table1.c.myid, table1.c.name]),
-            "SELECT mytable.myid, mytable.name, mytable.description "
-            "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE OF "
-            "mytable.myid, mytable.name NOWAIT")
-
-        self.assert_compile(
-            table1.select(table1.c.myid == 7)
-            .with_for_update(skip_locked=True,
-                             of=[table1.c.myid, table1.c.name]),
-            "SELECT mytable.myid, mytable.name, mytable.description "
-            "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE OF "
-            "mytable.myid, mytable.name SKIP LOCKED")
-
-        # key_share has no effect
-        self.assert_compile(
-            table1.select(table1.c.myid == 7).with_for_update(key_share=True),
-            "SELECT mytable.myid, mytable.name, mytable.description "
-            "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE")
-
-        # read has no effect
-        self.assert_compile(
-            table1
-            .select(table1.c.myid == 7)
-            .with_for_update(read=True, key_share=True),
-            "SELECT mytable.myid, mytable.name, mytable.description "
-            "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE")
-
-        ta = table1.alias()
-        self.assert_compile(
-            ta
-            .select(ta.c.myid == 7)
-            .with_for_update(of=[ta.c.myid, ta.c.name]),
-            "SELECT mytable_1.myid, mytable_1.name, mytable_1.description "
-            "FROM mytable mytable_1 "
-            "WHERE mytable_1.myid = :myid_1 FOR UPDATE OF "
-            "mytable_1.myid, mytable_1.name"
-        )
-
-    def test_for_update_of_w_limit_adaption_col_present(self):
-        table1 = table('mytable', column('myid'), column('name'))
-
-        self.assert_compile(
-            select([table1.c.myid, table1.c.name]).
-            where(table1.c.myid == 7).
-            with_for_update(nowait=True, of=table1.c.name).
-            limit(10),
-            "SELECT myid, name FROM "
-            "(SELECT mytable.myid AS myid, mytable.name AS name "
-            "FROM mytable WHERE mytable.myid = :myid_1) "
-            "WHERE ROWNUM <= :param_1 FOR UPDATE OF name NOWAIT",
-        )
-
-    def test_for_update_of_w_limit_adaption_col_unpresent(self):
-        table1 = table('mytable', column('myid'), column('name'))
-
-        self.assert_compile(
-            select([table1.c.myid]).
-            where(table1.c.myid == 7).
-            with_for_update(nowait=True, of=table1.c.name).
-            limit(10),
-            "SELECT myid FROM "
-            "(SELECT mytable.myid AS myid, mytable.name AS name "
-            "FROM mytable WHERE mytable.myid = :myid_1) "
-            "WHERE ROWNUM <= :param_1 FOR UPDATE OF name NOWAIT",
-        )
-
-    def test_for_update_of_w_limit_offset_adaption_col_present(self):
-        table1 = table('mytable', column('myid'), column('name'))
-
-        self.assert_compile(
-            select([table1.c.myid, table1.c.name]).
-            where(table1.c.myid == 7).
-            with_for_update(nowait=True, of=table1.c.name).
-            limit(10).offset(50),
-            "SELECT myid, name FROM (SELECT myid, name, ROWNUM AS ora_rn "
-            "FROM (SELECT mytable.myid AS myid, mytable.name AS name "
-            "FROM mytable WHERE mytable.myid = :myid_1) "
-            "WHERE ROWNUM <= :param_1 + :param_2) WHERE ora_rn > :param_2 "
-            "FOR UPDATE OF name NOWAIT",
-        )
-
-    def test_for_update_of_w_limit_offset_adaption_col_unpresent(self):
-        table1 = table('mytable', column('myid'), column('name'))
-
-        self.assert_compile(
-            select([table1.c.myid]).
-            where(table1.c.myid == 7).
-            with_for_update(nowait=True, of=table1.c.name).
-            limit(10).offset(50),
-            "SELECT myid FROM (SELECT myid, ROWNUM AS ora_rn, name "
-            "FROM (SELECT mytable.myid AS myid, mytable.name AS name "
-            "FROM mytable WHERE mytable.myid = :myid_1) "
-            "WHERE ROWNUM <= :param_1 + :param_2) WHERE ora_rn > :param_2 "
-            "FOR UPDATE OF name NOWAIT",
-        )
-
-    def test_for_update_of_w_limit_offset_adaption_partial_col_unpresent(self):
-        table1 = table('mytable', column('myid'), column('foo'), column('bar'))
-
-        self.assert_compile(
-            select([table1.c.myid, table1.c.bar]).
-            where(table1.c.myid == 7).
-            with_for_update(nowait=True, of=[table1.c.foo, table1.c.bar]).
-            limit(10).offset(50),
-            "SELECT myid, bar FROM (SELECT myid, bar, ROWNUM AS ora_rn, "
-            "foo FROM (SELECT mytable.myid AS myid, mytable.bar AS bar, "
-            "mytable.foo AS foo FROM mytable WHERE mytable.myid = :myid_1) "
-            "WHERE ROWNUM <= :param_1 + :param_2) WHERE ora_rn > :param_2 "
-            "FOR UPDATE OF foo, bar NOWAIT"
-        )
-
-    def test_limit_preserves_typing_information(self):
-        class MyType(TypeDecorator):
-            impl = Integer
-
-        stmt = select([type_coerce(column('x'), MyType).label('foo')]).limit(1)
-        dialect = oracle.dialect()
-        compiled = stmt.compile(dialect=dialect)
-        assert isinstance(compiled._create_result_map()['foo'][-1], MyType)
-
-    def test_use_binds_for_limits_disabled(self):
-        t = table('sometable', column('col1'), column('col2'))
-        dialect = oracle.OracleDialect(use_binds_for_limits=False)
-
-        self.assert_compile(
-            select([t]).limit(10),
-            "SELECT col1, col2 FROM (SELECT sometable.col1 AS col1, "
-            "sometable.col2 AS col2 FROM sometable) WHERE ROWNUM <= 10",
-            dialect=dialect)
-
-        self.assert_compile(
-            select([t]).offset(10),
-            "SELECT col1, col2 FROM (SELECT col1, col2, ROWNUM AS ora_rn "
-            "FROM (SELECT sometable.col1 AS col1, sometable.col2 AS col2 "
-            "FROM sometable)) WHERE ora_rn > 10",
-            dialect=dialect)
-
-        self.assert_compile(
-            select([t]).limit(10).offset(10),
-            "SELECT col1, col2 FROM (SELECT col1, col2, ROWNUM AS ora_rn "
-            "FROM (SELECT sometable.col1 AS col1, sometable.col2 AS col2 "
-            "FROM sometable) WHERE ROWNUM <= 20) WHERE ora_rn > 10",
-            dialect=dialect)
-
-    def test_use_binds_for_limits_enabled(self):
-        t = table('sometable', column('col1'), column('col2'))
-        dialect = oracle.OracleDialect(use_binds_for_limits=True)
-
-        self.assert_compile(
-            select([t]).limit(10),
-            "SELECT col1, col2 FROM (SELECT sometable.col1 AS col1, "
-            "sometable.col2 AS col2 FROM sometable) WHERE ROWNUM "
-            "<= :param_1",
-            dialect=dialect)
-
-        self.assert_compile(
-            select([t]).offset(10),
-            "SELECT col1, col2 FROM (SELECT col1, col2, ROWNUM AS ora_rn "
-            "FROM (SELECT sometable.col1 AS col1, sometable.col2 AS col2 "
-            "FROM sometable)) WHERE ora_rn > :param_1",
-            dialect=dialect)
-
-        self.assert_compile(
-            select([t]).limit(10).offset(10),
-            "SELECT col1, col2 FROM (SELECT col1, col2, ROWNUM AS ora_rn "
-            "FROM (SELECT sometable.col1 AS col1, sometable.col2 AS col2 "
-            "FROM sometable) WHERE ROWNUM <= :param_1 + :param_2) "
-            "WHERE ora_rn > :param_2",
-            dialect=dialect,
-            checkparams={'param_1': 10, 'param_2': 10})
-
-    def test_long_labels(self):
-        dialect = default.DefaultDialect()
-        dialect.max_identifier_length = 30
-
-        ora_dialect = oracle.dialect()
-
-        m = MetaData()
-        a_table = Table(
-            'thirty_characters_table_xxxxxx',
-            m,
-            Column('id', Integer, primary_key=True)
-        )
-
-        other_table = Table(
-            'other_thirty_characters_table_',
-            m,
-            Column('id', Integer, primary_key=True),
-            Column('thirty_characters_table_id',
-                   Integer,
-                   ForeignKey('thirty_characters_table_xxxxxx.id'),
-                   primary_key=True))
-
-        anon = a_table.alias()
-        self.assert_compile(select([other_table,
-                            anon]).
-                            select_from(
-                                other_table.outerjoin(anon)).apply_labels(),
-                            'SELECT other_thirty_characters_table_.id '
-                            'AS other_thirty_characters__1, '
-                            'other_thirty_characters_table_.thirty_char'
-                            'acters_table_id AS other_thirty_characters'
-                            '__2, thirty_characters_table__1.id AS '
-                            'thirty_characters_table__3 FROM '
-                            'other_thirty_characters_table_ LEFT OUTER '
-                            'JOIN thirty_characters_table_xxxxxx AS '
-                            'thirty_characters_table__1 ON '
-                            'thirty_characters_table__1.id = '
-                            'other_thirty_characters_table_.thirty_char'
-                            'acters_table_id', dialect=dialect)
-        self.assert_compile(select([other_table,
-                            anon]).select_from(
-                                other_table.outerjoin(anon)).apply_labels(),
-                            'SELECT other_thirty_characters_table_.id '
-                            'AS other_thirty_characters__1, '
-                            'other_thirty_characters_table_.thirty_char'
-                            'acters_table_id AS other_thirty_characters'
-                            '__2, thirty_characters_table__1.id AS '
-                            'thirty_characters_table__3 FROM '
-                            'other_thirty_characters_table_ LEFT OUTER '
-                            'JOIN thirty_characters_table_xxxxxx '
-                            'thirty_characters_table__1 ON '
-                            'thirty_characters_table__1.id = '
-                            'other_thirty_characters_table_.thirty_char'
-                            'acters_table_id', dialect=ora_dialect)
-
-    def test_outer_join(self):
-        table1 = table('mytable',
-                       column('myid', Integer),
-                       column('name', String),
-                       column('description', String))
-
-        table2 = table(
-            'myothertable',
-            column('otherid', Integer),
-            column('othername', String),
-        )
-
-        table3 = table(
-            'thirdtable',
-            column('userid', Integer),
-            column('otherstuff', String),
-        )
-
-        query = select([table1, table2],
-                       or_(table1.c.name == 'fred',
-                           table1.c.myid == 10, table2.c.othername != 'jack',
-                           text('EXISTS (select yay from foo where boo = lar)')
-                           ),
-                       from_obj=[outerjoin(table1,
-                                           table2,
-                                           table1.c.myid == table2.c.otherid)])
-        self.assert_compile(query,
-                            'SELECT mytable.myid, mytable.name, '
-                            'mytable.description, myothertable.otherid,'
-                            ' myothertable.othername FROM mytable, '
-                            'myothertable WHERE (mytable.name = '
-                            ':name_1 OR mytable.myid = :myid_1 OR '
-                            'myothertable.othername != :othername_1 OR '
-                            'EXISTS (select yay from foo where boo = '
-                            'lar)) AND mytable.myid = '
-                            'myothertable.otherid(+)',
-                            dialect=oracle.OracleDialect(use_ansi=False))
-        query = table1.outerjoin(table2,
-                                 table1.c.myid == table2.c.otherid) \
-                      .outerjoin(table3, table3.c.userid == table2.c.otherid)
-        self.assert_compile(query.select(),
-                            'SELECT mytable.myid, mytable.name, '
-                            'mytable.description, myothertable.otherid,'
-                            ' myothertable.othername, '
-                            'thirdtable.userid, thirdtable.otherstuff '
-                            'FROM mytable LEFT OUTER JOIN myothertable '
-                            'ON mytable.myid = myothertable.otherid '
-                            'LEFT OUTER JOIN thirdtable ON '
-                            'thirdtable.userid = myothertable.otherid')
-
-        self.assert_compile(query.select(),
-                            'SELECT mytable.myid, mytable.name, '
-                            'mytable.description, myothertable.otherid,'
-                            ' myothertable.othername, '
-                            'thirdtable.userid, thirdtable.otherstuff '
-                            'FROM mytable, myothertable, thirdtable '
-                            'WHERE thirdtable.userid(+) = '
-                            'myothertable.otherid AND mytable.myid = '
-                            'myothertable.otherid(+)',
-                            dialect=oracle.dialect(use_ansi=False))
-        query = table1.join(table2,
-                            table1.c.myid == table2.c.otherid) \
-                      .join(table3, table3.c.userid == table2.c.otherid)
-        self.assert_compile(query.select(),
-                            'SELECT mytable.myid, mytable.name, '
-                            'mytable.description, myothertable.otherid,'
-                            ' myothertable.othername, '
-                            'thirdtable.userid, thirdtable.otherstuff '
-                            'FROM mytable, myothertable, thirdtable '
-                            'WHERE thirdtable.userid = '
-                            'myothertable.otherid AND mytable.myid = '
-                            'myothertable.otherid',
-                            dialect=oracle.dialect(use_ansi=False))
-        query = table1.join(table2,
-                            table1.c.myid == table2.c.otherid) \
-                      .outerjoin(table3, table3.c.userid == table2.c.otherid)
-        self.assert_compile(query.select().order_by(table1.c.name).
-                            limit(10).offset(5),
-                            'SELECT myid, name, description, otherid, '
-                            'othername, userid, otherstuff FROM '
-                            '(SELECT myid, name, description, otherid, '
-                            'othername, userid, otherstuff, ROWNUM AS '
-                            'ora_rn FROM (SELECT mytable.myid AS myid, '
-                            'mytable.name AS name, mytable.description '
-                            'AS description, myothertable.otherid AS '
-                            'otherid, myothertable.othername AS '
-                            'othername, thirdtable.userid AS userid, '
-                            'thirdtable.otherstuff AS otherstuff FROM '
-                            'mytable, myothertable, thirdtable WHERE '
-                            'thirdtable.userid(+) = '
-                            'myothertable.otherid AND mytable.myid = '
-                            'myothertable.otherid ORDER BY mytable.name) '
-                            'WHERE ROWNUM <= :param_1 + :param_2) '
-                            'WHERE ora_rn > :param_2',
-                            checkparams={'param_1': 10, 'param_2': 5},
-                            dialect=oracle.dialect(use_ansi=False))
-
-        subq = select([table1]).select_from(
-            table1.outerjoin(table2, table1.c.myid == table2.c.otherid)) \
-            .alias()
-        q = select([table3]).select_from(
-            table3.outerjoin(subq, table3.c.userid == subq.c.myid))
-
-        self.assert_compile(q,
-                            'SELECT thirdtable.userid, '
-                            'thirdtable.otherstuff FROM thirdtable '
-                            'LEFT OUTER JOIN (SELECT mytable.myid AS '
-                            'myid, mytable.name AS name, '
-                            'mytable.description AS description FROM '
-                            'mytable LEFT OUTER JOIN myothertable ON '
-                            'mytable.myid = myothertable.otherid) '
-                            'anon_1 ON thirdtable.userid = anon_1.myid',
-                            dialect=oracle.dialect(use_ansi=True))
-
-        self.assert_compile(q,
-                            'SELECT thirdtable.userid, '
-                            'thirdtable.otherstuff FROM thirdtable, '
-                            '(SELECT mytable.myid AS myid, '
-                            'mytable.name AS name, mytable.description '
-                            'AS description FROM mytable, myothertable '
-                            'WHERE mytable.myid = myothertable.otherid('
-                            '+)) anon_1 WHERE thirdtable.userid = '
-                            'anon_1.myid(+)',
-                            dialect=oracle.dialect(use_ansi=False))
-
-        q = select([table1.c.name]).where(table1.c.name == 'foo')
-        self.assert_compile(q,
-                            'SELECT mytable.name FROM mytable WHERE '
-                            'mytable.name = :name_1',
-                            dialect=oracle.dialect(use_ansi=False))
-        subq = select([table3.c.otherstuff]) \
-            .where(table3.c.otherstuff == table1.c.name).label('bar')
-        q = select([table1.c.name, subq])
-        self.assert_compile(q,
-                            'SELECT mytable.name, (SELECT '
-                            'thirdtable.otherstuff FROM thirdtable '
-                            'WHERE thirdtable.otherstuff = '
-                            'mytable.name) AS bar FROM mytable',
-                            dialect=oracle.dialect(use_ansi=False))
-
-    def test_nonansi_nested_right_join(self):
-        a = table('a', column('a'))
-        b = table('b', column('b'))
-        c = table('c', column('c'))
-
-        j = a.join(b.join(c, b.c.b == c.c.c), a.c.a == b.c.b)
-
-        self.assert_compile(
-            select([j]),
-            "SELECT a.a, b.b, c.c FROM a, b, c "
-            "WHERE a.a = b.b AND b.b = c.c",
-            dialect=oracle.OracleDialect(use_ansi=False)
-        )
-
-        j = a.outerjoin(b.join(c, b.c.b == c.c.c), a.c.a == b.c.b)
-
-        self.assert_compile(
-            select([j]),
-            "SELECT a.a, b.b, c.c FROM a, b, c "
-            "WHERE a.a = b.b(+) AND b.b = c.c",
-            dialect=oracle.OracleDialect(use_ansi=False)
-        )
-
-        j = a.join(b.outerjoin(c, b.c.b == c.c.c), a.c.a == b.c.b)
-
-        self.assert_compile(
-            select([j]),
-            "SELECT a.a, b.b, c.c FROM a, b, c "
-            "WHERE a.a = b.b AND b.b = c.c(+)",
-            dialect=oracle.OracleDialect(use_ansi=False)
-        )
-
-    def test_alias_outer_join(self):
-        address_types = table('address_types', column('id'),
-                              column('name'))
-        addresses = table('addresses', column('id'), column('user_id'),
-                          column('address_type_id'),
-                          column('email_address'))
-        at_alias = address_types.alias()
-        s = select([at_alias, addresses]) \
-            .select_from(
-                addresses.outerjoin(
-                    at_alias,
-                    addresses.c.address_type_id == at_alias.c.id)) \
-            .where(addresses.c.user_id == 7) \
-            .order_by(addresses.c.id, address_types.c.id)
-        self.assert_compile(s,
-                            'SELECT address_types_1.id, '
-                            'address_types_1.name, addresses.id, '
-                            'addresses.user_id, addresses.address_type_'
-                            'id, addresses.email_address FROM '
-                            'addresses LEFT OUTER JOIN address_types '
-                            'address_types_1 ON addresses.address_type_'
-                            'id = address_types_1.id WHERE '
-                            'addresses.user_id = :user_id_1 ORDER BY '
-                            'addresses.id, address_types.id')
-
-    def test_returning_insert(self):
-        t1 = table('t1', column('c1'), column('c2'), column('c3'))
-        self.assert_compile(
-            t1.insert().values(c1=1).returning(t1.c.c2, t1.c.c3),
-            "INSERT INTO t1 (c1) VALUES (:c1) RETURNING "
-            "t1.c2, t1.c3 INTO :ret_0, :ret_1")
-
-    def test_returning_insert_functional(self):
-        t1 = table('t1',
-                   column('c1'),
-                   column('c2', String()),
-                   column('c3', String()))
-        fn = func.lower(t1.c.c2, type_=String())
-        stmt = t1.insert().values(c1=1).returning(fn, t1.c.c3)
-        compiled = stmt.compile(dialect=oracle.dialect())
-        eq_(compiled._create_result_map(),
-            {'ret_1': ('ret_1', (t1.c.c3, 'c3', 'c3'), t1.c.c3.type),
-            'ret_0': ('ret_0', (fn, 'lower', None), fn.type)})
-        self.assert_compile(
-            stmt,
-            "INSERT INTO t1 (c1) VALUES (:c1) RETURNING "
-            "lower(t1.c2), t1.c3 INTO :ret_0, :ret_1")
-
-    def test_returning_insert_labeled(self):
-        t1 = table('t1', column('c1'), column('c2'), column('c3'))
-        self.assert_compile(
-            t1.insert().values(c1=1).returning(
-                        t1.c.c2.label('c2_l'), t1.c.c3.label('c3_l')),
-            "INSERT INTO t1 (c1) VALUES (:c1) RETURNING "
-            "t1.c2, t1.c3 INTO :ret_0, :ret_1")
-
-    def test_compound(self):
-        t1 = table('t1', column('c1'), column('c2'), column('c3'))
-        t2 = table('t2', column('c1'), column('c2'), column('c3'))
-        self.assert_compile(union(t1.select(), t2.select()),
-                            'SELECT t1.c1, t1.c2, t1.c3 FROM t1 UNION '
-                            'SELECT t2.c1, t2.c2, t2.c3 FROM t2')
-        self.assert_compile(except_(t1.select(), t2.select()),
-                            'SELECT t1.c1, t1.c2, t1.c3 FROM t1 MINUS '
-                            'SELECT t2.c1, t2.c2, t2.c3 FROM t2')
-
-    def test_no_paren_fns(self):
-        for fn, expected in [
-            (func.uid(), "uid"),
-            (func.UID(), "UID"),
-            (func.sysdate(), "sysdate"),
-            (func.row_number(), "row_number()"),
-            (func.rank(), "rank()"),
-            (func.now(), "CURRENT_TIMESTAMP"),
-            (func.current_timestamp(), "CURRENT_TIMESTAMP"),
-            (func.user(), "USER"),
-        ]:
-            self.assert_compile(fn, expected)
-
-    def test_create_index_alt_schema(self):
-        m = MetaData()
-        t1 = Table('foo', m,
-                   Column('x', Integer),
-                   schema="alt_schema")
-        self.assert_compile(
-            schema.CreateIndex(Index("bar", t1.c.x)),
-            "CREATE INDEX alt_schema.bar ON alt_schema.foo (x)"
-        )
-
-    def test_create_index_expr(self):
-        m = MetaData()
-        t1 = Table('foo', m,
-                   Column('x', Integer))
-        self.assert_compile(
-            schema.CreateIndex(Index("bar", t1.c.x > 5)),
-            "CREATE INDEX bar ON foo (x > 5)"
-        )
-
-    def test_table_options(self):
-        m = MetaData()
-
-        t = Table(
-            'foo', m,
-            Column('x', Integer),
-            prefixes=["GLOBAL TEMPORARY"],
-            oracle_on_commit="PRESERVE ROWS"
-        )
-
-        self.assert_compile(
-            schema.CreateTable(t),
-            "CREATE GLOBAL TEMPORARY TABLE "
-            "foo (x INTEGER) ON COMMIT PRESERVE ROWS"
-        )
-
-    def test_create_table_compress(self):
-        m = MetaData()
-        tbl1 = Table('testtbl1', m, Column('data', Integer),
-                     oracle_compress=True)
-        tbl2 = Table('testtbl2', m, Column('data', Integer),
-                     oracle_compress="OLTP")
-
-        self.assert_compile(schema.CreateTable(tbl1),
-                            "CREATE TABLE testtbl1 (data INTEGER) COMPRESS")
-        self.assert_compile(schema.CreateTable(tbl2),
-                            "CREATE TABLE testtbl2 (data INTEGER) "
-                            "COMPRESS FOR OLTP")
-
-    def test_create_index_bitmap_compress(self):
-        m = MetaData()
-        tbl = Table('testtbl', m, Column('data', Integer))
-        idx1 = Index('idx1', tbl.c.data, oracle_compress=True)
-        idx2 = Index('idx2', tbl.c.data, oracle_compress=1)
-        idx3 = Index('idx3', tbl.c.data, oracle_bitmap=True)
-
-        self.assert_compile(schema.CreateIndex(idx1),
-                            "CREATE INDEX idx1 ON testtbl (data) COMPRESS")
-        self.assert_compile(schema.CreateIndex(idx2),
-                            "CREATE INDEX idx2 ON testtbl (data) COMPRESS 1")
-        self.assert_compile(schema.CreateIndex(idx3),
-                            "CREATE BITMAP INDEX idx3 ON testtbl (data)")
-
-
-class CompatFlagsTest(fixtures.TestBase, AssertsCompiledSQL):
-
-    def _dialect(self, server_version, **kw):
-        def server_version_info(conn):
-            return server_version
-
-        dialect = oracle.dialect(
-                        dbapi=Mock(version="0.0.0", paramstyle="named"),
-                        **kw)
-        dialect._get_server_version_info = server_version_info
-        dialect._check_unicode_returns = Mock()
-        dialect._check_unicode_description = Mock()
-        dialect._get_default_schema_name = Mock()
-        return dialect
-
-    def test_ora8_flags(self):
-        dialect = self._dialect((8, 2, 5))
-
-        # before connect, assume modern DB
-        assert dialect._supports_char_length
-        assert dialect._supports_nchar
-        assert dialect.use_ansi
-
-        dialect.initialize(Mock())
-        assert not dialect.implicit_returning
-        assert not dialect._supports_char_length
-        assert not dialect._supports_nchar
-        assert not dialect.use_ansi
-        self.assert_compile(String(50), "VARCHAR2(50)", dialect=dialect)
-        self.assert_compile(Unicode(50), "VARCHAR2(50)", dialect=dialect)
-        self.assert_compile(UnicodeText(), "CLOB", dialect=dialect)
-
-        dialect = self._dialect((8, 2, 5), implicit_returning=True)
-        dialect.initialize(testing.db.connect())
-        assert dialect.implicit_returning
-
-    def test_default_flags(self):
-        """test with no initialization or server version info"""
-
-        dialect = self._dialect(None)
-
-        assert dialect._supports_char_length
-        assert dialect._supports_nchar
-        assert dialect.use_ansi
-        self.assert_compile(String(50), "VARCHAR2(50 CHAR)", dialect=dialect)
-        self.assert_compile(Unicode(50), "NVARCHAR2(50)", dialect=dialect)
-        self.assert_compile(UnicodeText(), "NCLOB", dialect=dialect)
-
-    def test_ora10_flags(self):
-        dialect = self._dialect((10, 2, 5))
-
-        dialect.initialize(Mock())
-        assert dialect._supports_char_length
-        assert dialect._supports_nchar
-        assert dialect.use_ansi
-        self.assert_compile(String(50), "VARCHAR2(50 CHAR)", dialect=dialect)
-        self.assert_compile(Unicode(50), "NVARCHAR2(50)", dialect=dialect)
-        self.assert_compile(UnicodeText(), "NCLOB", dialect=dialect)
-
-
-class MultiSchemaTest(fixtures.TestBase, AssertsCompiledSQL):
-    __only_on__ = 'oracle'
-    __backend__ = True
-
-    @classmethod
-    def setup_class(cls):
-        # currently assuming full DBA privs for the user.
-        # don't really know how else to go here unless
-        # we connect as the other user.
-
-        for stmt in ("""
-create table %(test_schema)s.parent(
-    id integer primary key,
-    data varchar2(50)
-);
-
-create table %(test_schema)s.child(
-    id integer primary key,
-    data varchar2(50),
-    parent_id integer references %(test_schema)s.parent(id)
-);
-
-create table local_table(
-    id integer primary key,
-    data varchar2(50)
-);
-
-create synonym %(test_schema)s.ptable for %(test_schema)s.parent;
-create synonym %(test_schema)s.ctable for %(test_schema)s.child;
-
-create synonym %(test_schema)s_pt for %(test_schema)s.parent;
-
-create synonym %(test_schema)s.local_table for local_table;
-
--- can't make a ref from local schema to the
--- remote schema's table without this,
--- *and* cant give yourself a grant !
--- so we give it to public.  ideas welcome.
-grant references on %(test_schema)s.parent to public;
-grant references on %(test_schema)s.child to public;
-""" % {"test_schema": testing.config.test_schema}).split(";"):
-            if stmt.strip():
-                testing.db.execute(stmt)
-
-    @classmethod
-    def teardown_class(cls):
-        for stmt in ("""
-drop table %(test_schema)s.child;
-drop table %(test_schema)s.parent;
-drop table local_table;
-drop synonym %(test_schema)s.ctable;
-drop synonym %(test_schema)s.ptable;
-drop synonym %(test_schema)s_pt;
-drop synonym %(test_schema)s.local_table;
-
-""" % {"test_schema": testing.config.test_schema}).split(";"):
-            if stmt.strip():
-                testing.db.execute(stmt)
-
-    @testing.provide_metadata
-    def test_create_same_names_explicit_schema(self):
-        schema = testing.db.dialect.default_schema_name
-        meta = self.metadata
-        parent = Table('parent', meta,
-                       Column('pid', Integer, primary_key=True),
-                       schema=schema)
-        child = Table('child', meta,
-                      Column('cid', Integer, primary_key=True),
-                      Column('pid',
-                             Integer,
-                             ForeignKey('%s.parent.pid' % schema)),
-                      schema=schema)
-        meta.create_all()
-        parent.insert().execute({'pid': 1})
-        child.insert().execute({'cid': 1, 'pid': 1})
-        eq_(child.select().execute().fetchall(), [(1, 1)])
-
-    def test_reflect_alt_table_owner_local_synonym(self):
-        meta = MetaData(testing.db)
-        parent = Table('%s_pt' % testing.config.test_schema,
-                       meta,
-                       autoload=True,
-                       oracle_resolve_synonyms=True)
-        self.assert_compile(parent.select(),
-                            "SELECT %(test_schema)s_pt.id, "
-                            "%(test_schema)s_pt.data FROM %(test_schema)s_pt"
-                            % {"test_schema": testing.config.test_schema})
-        select([parent]).execute().fetchall()
-
-    def test_reflect_alt_synonym_owner_local_table(self):
-        meta = MetaData(testing.db)
-        parent = Table(
-            'local_table', meta, autoload=True,
-            oracle_resolve_synonyms=True, schema=testing.config.test_schema)
-        self.assert_compile(
-            parent.select(),
-            "SELECT %(test_schema)s.local_table.id, "
-            "%(test_schema)s.local_table.data "
-            "FROM %(test_schema)s.local_table" %
-            {"test_schema": testing.config.test_schema}
-        )
-        select([parent]).execute().fetchall()
-
-    @testing.provide_metadata
-    def test_create_same_names_implicit_schema(self):
-        meta = self.metadata
-        parent = Table('parent',
-                       meta,
-                       Column('pid', Integer, primary_key=True))
-        child = Table('child', meta,
-                      Column('cid', Integer, primary_key=True),
-                      Column('pid', Integer, ForeignKey('parent.pid')))
-        meta.create_all()
-        parent.insert().execute({'pid': 1})
-        child.insert().execute({'cid': 1, 'pid': 1})
-        eq_(child.select().execute().fetchall(), [(1, 1)])
-
-    def test_reflect_alt_owner_explicit(self):
-        meta = MetaData(testing.db)
-        parent = Table(
-            'parent', meta, autoload=True,
-            schema=testing.config.test_schema)
-        child = Table(
-            'child', meta, autoload=True,
-            schema=testing.config.test_schema)
-
-        self.assert_compile(
-            parent.join(child),
-            "%(test_schema)s.parent JOIN %(test_schema)s.child ON "
-            "%(test_schema)s.parent.id = %(test_schema)s.child.parent_id" % {
-                "test_schema": testing.config.test_schema
-            })
-        select([parent, child]).\
-            select_from(parent.join(child)).\
-            execute().fetchall()
-
-    def test_reflect_local_to_remote(self):
-        testing.db.execute(
-            'CREATE TABLE localtable (id INTEGER '
-            'PRIMARY KEY, parent_id INTEGER REFERENCES '
-            '%(test_schema)s.parent(id))' % {
-                "test_schema": testing.config.test_schema})
-        try:
-            meta = MetaData(testing.db)
-            lcl = Table('localtable', meta, autoload=True)
-            parent = meta.tables['%s.parent' % testing.config.test_schema]
-            self.assert_compile(parent.join(lcl),
-                                '%(test_schema)s.parent JOIN localtable ON '
-                                '%(test_schema)s.parent.id = '
-                                'localtable.parent_id' % {
-                                    "test_schema": testing.config.test_schema}
-                                )
-            select([parent,
-                   lcl]).select_from(parent.join(lcl)).execute().fetchall()
-        finally:
-            testing.db.execute('DROP TABLE localtable')
-
-    def test_reflect_alt_owner_implicit(self):
-        meta = MetaData(testing.db)
-        parent = Table(
-            'parent', meta, autoload=True,
-            schema=testing.config.test_schema)
-        child = Table(
-            'child', meta, autoload=True,
-            schema=testing.config.test_schema)
-        self.assert_compile(
-            parent.join(child),
-            '%(test_schema)s.parent JOIN %(test_schema)s.child '
-            'ON %(test_schema)s.parent.id = '
-            '%(test_schema)s.child.parent_id' % {
-                "test_schema": testing.config.test_schema})
-        select([parent,
-               child]).select_from(parent.join(child)).execute().fetchall()
-
-    def test_reflect_alt_owner_synonyms(self):
-        testing.db.execute('CREATE TABLE localtable (id INTEGER '
-                           'PRIMARY KEY, parent_id INTEGER REFERENCES '
-                           '%s.ptable(id))' % testing.config.test_schema)
-        try:
-            meta = MetaData(testing.db)
-            lcl = Table('localtable', meta, autoload=True,
-                        oracle_resolve_synonyms=True)
-            parent = meta.tables['%s.ptable' % testing.config.test_schema]
-            self.assert_compile(
-                parent.join(lcl),
-                '%(test_schema)s.ptable JOIN localtable ON '
-                '%(test_schema)s.ptable.id = '
-                'localtable.parent_id' % {
-                    "test_schema": testing.config.test_schema})
-            select([parent,
-                   lcl]).select_from(parent.join(lcl)).execute().fetchall()
-        finally:
-            testing.db.execute('DROP TABLE localtable')
-
-    def test_reflect_remote_synonyms(self):
-        meta = MetaData(testing.db)
-        parent = Table('ptable', meta, autoload=True,
-                       schema=testing.config.test_schema,
-                       oracle_resolve_synonyms=True)
-        child = Table('ctable', meta, autoload=True,
-                      schema=testing.config.test_schema,
-                      oracle_resolve_synonyms=True)
-        self.assert_compile(
-            parent.join(child),
-            '%(test_schema)s.ptable JOIN '
-            '%(test_schema)s.ctable '
-            'ON %(test_schema)s.ptable.id = '
-            '%(test_schema)s.ctable.parent_id' % {
-                "test_schema": testing.config.test_schema})
-        select([parent,
-               child]).select_from(parent.join(child)).execute().fetchall()
-
-
-class ConstraintTest(fixtures.TablesTest):
-
-    __only_on__ = 'oracle'
-    __backend__ = True
-    run_deletes = None
-
-    @classmethod
-    def define_tables(cls, metadata):
-        Table('foo', metadata, Column('id', Integer, primary_key=True))
-
-    def test_oracle_has_no_on_update_cascade(self):
-        bar = Table('bar', self.metadata,
-                    Column('id', Integer, primary_key=True),
-                    Column('foo_id',
-                           Integer,
-                           ForeignKey('foo.id', onupdate='CASCADE')))
-        assert_raises(exc.SAWarning, bar.create)
-
-        bat = Table('bat', self.metadata,
-                    Column('id', Integer, primary_key=True),
-                    Column('foo_id', Integer),
-                    ForeignKeyConstraint(['foo_id'], ['foo.id'],
-                                         onupdate='CASCADE'))
-        assert_raises(exc.SAWarning, bat.create)
-
-    def test_reflect_check_include_all(self):
-        insp = inspect(testing.db)
-        eq_(insp.get_check_constraints('foo'), [])
-        eq_(
-            [rec['sqltext']
-             for rec in insp.get_check_constraints('foo', include_all=True)],
-            ['"ID" IS NOT NULL'])
-
-
-class TwoPhaseTest(fixtures.TablesTest):
-    """test cx_oracle two phase, which remains in a semi-broken state
-    so requires a carefully written test."""
-
-    __only_on__ = 'oracle+cx_oracle'
-    __backend__ = True
-
-    @classmethod
-    def define_tables(cls, metadata):
-        Table('datatable', metadata,
-              Column('id', Integer, primary_key=True),
-              Column('data', String(50)))
-
-    def _connection(self):
-        conn = testing.db.connect()
-        conn.detach()
-        return conn
-
-    def _assert_data(self, rows):
-        eq_(
-            testing.db.scalar("select count(*) from datatable"),
-            rows
-        )
-
-    def test_twophase_prepare_false(self):
-        conn = self._connection()
-        for i in range(2):
-            trans = conn.begin_twophase()
-            conn.execute("select 1 from dual")
-            trans.prepare()
-            trans.commit()
-        conn.close()
-        self._assert_data(0)
-
-    def test_twophase_prepare_true(self):
-        conn = self._connection()
-        for i in range(2):
-            trans = conn.begin_twophase()
-            conn.execute("insert into datatable (id, data) "
-                         "values (%s, 'somedata')" % i)
-            trans.prepare()
-            trans.commit()
-        conn.close()
-        self._assert_data(2)
-
-    def test_twophase_rollback(self):
-        conn = self._connection()
-        trans = conn.begin_twophase()
-        conn.execute("insert into datatable (id, data) "
-                     "values (%s, 'somedata')" % 1)
-        trans.rollback()
-
-        trans = conn.begin_twophase()
-        conn.execute("insert into datatable (id, data) "
-                     "values (%s, 'somedata')" % 1)
-        trans.prepare()
-        trans.commit()
-
-        conn.close()
-        self._assert_data(1)
-
-    def test_not_prepared(self):
-        conn = self._connection()
-        trans = conn.begin_twophase()
-        conn.execute("insert into datatable (id, data) "
-                     "values (%s, 'somedata')" % 1)
-        trans.commit()
-        conn.close()
-        self._assert_data(1)
-
-
-class DialectTypesTest(fixtures.TestBase, AssertsCompiledSQL):
-    __dialect__ = oracle.OracleDialect()
-
-    def test_no_clobs_for_string_params(self):
-        """test that simple string params get a DBAPI type of
-        VARCHAR, not CLOB. This is to prevent setinputsizes
-        from setting up cx_oracle.CLOBs on
-        string-based bind params [ticket:793]."""
-
-        class FakeDBAPI(object):
-            def __getattr__(self, attr):
-                return attr
-
-        dialect = oracle.OracleDialect()
-        dbapi = FakeDBAPI()
-
-        b = bindparam("foo", "hello world!")
-        eq_(
-            b.type.dialect_impl(dialect).get_dbapi_type(dbapi),
-            'STRING'
-        )
-
-        b = bindparam("foo", "hello world!")
-        eq_(
-            b.type.dialect_impl(dialect).get_dbapi_type(dbapi),
-            'STRING'
-        )
-
-    def test_long(self):
-        self.assert_compile(oracle.LONG(), "LONG")
-
-    def test_type_adapt(self):
-        dialect = cx_oracle.dialect()
-
-        for start, test in [
-            (Date(), cx_oracle._OracleDate),
-            (oracle.OracleRaw(), cx_oracle._OracleRaw),
-            (String(), String),
-            (VARCHAR(), cx_oracle._OracleString),
-            (DATE(), cx_oracle._OracleDate),
-            (oracle.DATE(), oracle.DATE),
-            (String(50), cx_oracle._OracleString),
-            (Unicode(), cx_oracle._OracleNVarChar),
-            (Text(), cx_oracle._OracleText),
-            (UnicodeText(), cx_oracle._OracleUnicodeText),
-            (NCHAR(), cx_oracle._OracleNVarChar),
-            (oracle.RAW(50), cx_oracle._OracleRaw),
-        ]:
-            assert isinstance(start.dialect_impl(dialect), test), \
-                    "wanted %r got %r" % (test, start.dialect_impl(dialect))
-
-    def test_raw_compile(self):
-        self.assert_compile(oracle.RAW(), "RAW")
-        self.assert_compile(oracle.RAW(35), "RAW(35)")
-
-    def test_char_length(self):
-        self.assert_compile(VARCHAR(50), "VARCHAR(50 CHAR)")
-
-        oracle8dialect = oracle.dialect()
-        oracle8dialect.server_version_info = (8, 0)
-        self.assert_compile(VARCHAR(50), "VARCHAR(50)", dialect=oracle8dialect)
-
-        self.assert_compile(NVARCHAR(50), "NVARCHAR2(50)")
-        self.assert_compile(CHAR(50), "CHAR(50)")
-
-    def test_varchar_types(self):
-        dialect = oracle.dialect()
-        for typ, exp in [
-            (String(50), "VARCHAR2(50 CHAR)"),
-            (Unicode(50), "NVARCHAR2(50)"),
-            (NVARCHAR(50), "NVARCHAR2(50)"),
-            (VARCHAR(50), "VARCHAR(50 CHAR)"),
-            (oracle.NVARCHAR2(50), "NVARCHAR2(50)"),
-            (oracle.VARCHAR2(50), "VARCHAR2(50 CHAR)"),
-            (String(), "VARCHAR2"),
-            (Unicode(), "NVARCHAR2"),
-            (NVARCHAR(), "NVARCHAR2"),
-            (VARCHAR(), "VARCHAR"),
-            (oracle.NVARCHAR2(), "NVARCHAR2"),
-            (oracle.VARCHAR2(), "VARCHAR2"),
-        ]:
-            self.assert_compile(typ, exp, dialect=dialect)
-
-    def test_interval(self):
-        for type_, expected in [(oracle.INTERVAL(),
-                                'INTERVAL DAY TO SECOND'),
-                                (oracle.INTERVAL(day_precision=3),
-                                'INTERVAL DAY(3) TO SECOND'),
-                                (oracle.INTERVAL(second_precision=5),
-                                'INTERVAL DAY TO SECOND(5)'),
-                                (oracle.INTERVAL(day_precision=2,
-                                 second_precision=5),
-                                'INTERVAL DAY(2) TO SECOND(5)')]:
-            self.assert_compile(type_, expected)
-
-
-class TypesTest(fixtures.TestBase):
-    __only_on__ = 'oracle'
-    __dialect__ = oracle.OracleDialect()
-    __backend__ = True
-
-    @testing.fails_on('+zxjdbc', 'zxjdbc lacks the FIXED_CHAR dbapi type')
-    def test_fixed_char(self):
-        m = MetaData(testing.db)
-        t = Table('t1', m,
-                  Column('id', Integer, primary_key=True),
-                  Column('data', CHAR(30), nullable=False))
-
-        t.create()
-        try:
-            t.insert().execute(
-                dict(id=1, data="value 1"),
-                dict(id=2, data="value 2"),
-                dict(id=3, data="value 3")
-            )
-
-            eq_(
-                t.select().where(t.c.data == 'value 2').execute().fetchall(),
-                [(2, 'value 2                       ')]
-            )
-
-            m2 = MetaData(testing.db)
-            t2 = Table('t1', m2, autoload=True)
-            assert type(t2.c.data.type) is CHAR
-            eq_(
-                t2.select().where(t2.c.data == 'value 2').execute().fetchall(),
-                [(2, 'value 2                       ')]
-            )
-
-        finally:
-            t.drop()
-
-    @testing.requires.returning
-    @testing.provide_metadata
-    def test_int_not_float(self):
-        m = self.metadata
-        t1 = Table('t1', m, Column('foo', Integer))
-        t1.create()
-        r = t1.insert().values(foo=5).returning(t1.c.foo).execute()
-        x = r.scalar()
-        assert x == 5
-        assert isinstance(x, int)
-
-        x = t1.select().scalar()
-        assert x == 5
-        assert isinstance(x, int)
-
-    @testing.provide_metadata
-    def test_rowid(self):
-        metadata = self.metadata
-        t = Table('t1', metadata, Column('x', Integer))
-        t.create()
-        t.insert().execute(x=5)
-        s1 = select([t])
-        s2 = select([column('rowid')]).select_from(s1)
-        rowid = s2.scalar()
-
-        # the ROWID type is not really needed here,
-        # as cx_oracle just treats it as a string,
-        # but we want to make sure the ROWID works...
-        rowid_col = column('rowid', oracle.ROWID)
-        s3 = select([t.c.x, rowid_col]) \
-            .where(rowid_col == cast(rowid, oracle.ROWID))
-        eq_(s3.select().execute().fetchall(), [(5, rowid)])
-
-    @testing.fails_on('+zxjdbc',
-                      'Not yet known how to pass values of the '
-                      'INTERVAL type')
-    @testing.provide_metadata
-    def test_interval(self):
-        metadata = self.metadata
-        interval_table = Table('intervaltable', metadata, Column('id',
-                               Integer, primary_key=True,
-                               test_needs_autoincrement=True),
-                               Column('day_interval',
-                               oracle.INTERVAL(day_precision=3)))
-        metadata.create_all()
-        interval_table.insert().\
-            execute(day_interval=datetime.timedelta(days=35, seconds=5743))
-        row = interval_table.select().execute().first()
-        eq_(row['day_interval'], datetime.timedelta(days=35,
-            seconds=5743))
-
-    @testing.provide_metadata
-    def test_numerics(self):
-        m = self.metadata
-        t1 = Table('t1', m,
-                   Column('intcol', Integer),
-                   Column('numericcol', Numeric(precision=9, scale=2)),
-                   Column('floatcol1', Float()),
-                   Column('floatcol2', FLOAT()),
-                   Column('doubleprec', oracle.DOUBLE_PRECISION),
-                   Column('numbercol1', oracle.NUMBER(9)),
-                   Column('numbercol2', oracle.NUMBER(9, 3)),
-                   Column('numbercol3', oracle.NUMBER))
-        t1.create()
-        t1.insert().execute(
-            intcol=1,
-            numericcol=5.2,
-            floatcol1=6.5,
-            floatcol2=8.5,
-            doubleprec=9.5,
-            numbercol1=12,
-            numbercol2=14.85,
-            numbercol3=15.76
-            )
-
-        m2 = MetaData(testing.db)
-        t2 = Table('t1', m2, autoload=True)
-
-        for row in (
-            t1.select().execute().first(),
-            t2.select().execute().first()
-        ):
-            for i, (val, type_) in enumerate((
-                (1, int),
-                (decimal.Decimal("5.2"), decimal.Decimal),
-                (6.5, float),
-                (8.5, float),
-                (9.5, float),
-                (12, int),
-                (decimal.Decimal("14.85"), decimal.Decimal),
-                (15.76, float),
-            )):
-                eq_(row[i], val)
-                assert isinstance(row[i], type_), '%r is not %r' \
-                    % (row[i], type_)
-
-    def test_numeric_no_decimal_mode(self):
-        engine = testing_engine(options=dict(coerce_to_decimal=False))
-        value = engine.scalar("SELECT 5.66 FROM DUAL")
-        assert isinstance(value, float)
-
-        value = testing.db.scalar("SELECT 5.66 FROM DUAL")
-        assert isinstance(value, decimal.Decimal)
-
-    @testing.only_on("oracle+cx_oracle", "cx_oracle-specific feature")
-    @testing.fails_if(
-                    testing.requires.python3,
-                    "cx_oracle always returns unicode on py3k")
-    def test_coerce_to_unicode(self):
-        engine = testing_engine(options=dict(coerce_to_unicode=True))
-        value = engine.scalar("SELECT 'hello' FROM DUAL")
-        assert isinstance(value, util.text_type)
-
-        value = testing.db.scalar("SELECT 'hello' FROM DUAL")
-        assert isinstance(value, util.binary_type)
-
-    @testing.provide_metadata
-    def test_numerics_broken_inspection(self):
-        """Numeric scenarios where Oracle type info is 'broken',
-        returning us precision, scale of the form (0, 0) or (0, -127).
-        We convert to Decimal and let int()/float() processors take over.
-
-        """
-
-        metadata = self.metadata
-
-        # this test requires cx_oracle 5
-
-        foo = Table('foo', metadata,
-                    Column('idata', Integer),
-                    Column('ndata', Numeric(20, 2)),
-                    Column('ndata2', Numeric(20, 2)),
-                    Column('nidata', Numeric(5, 0)),
-                    Column('fdata', Float()))
-        foo.create()
-
-        foo.insert().execute({
-                'idata': 5,
-                'ndata': decimal.Decimal("45.6"),
-                'ndata2': decimal.Decimal("45.0"),
-                'nidata': decimal.Decimal('53'),
-                'fdata': 45.68392
-            })
-
-        stmt = "SELECT idata, ndata, ndata2, nidata, fdata FROM foo"
-
-        row = testing.db.execute(stmt).fetchall()[0]
-        eq_(
-            [type(x) for x in row],
-            [int, decimal.Decimal, decimal.Decimal, int, float]
-        )
-        eq_(
-            row,
-            (5, decimal.Decimal('45.6'), decimal.Decimal('45'),
-                53, 45.683920000000001)
-        )
-
-        # with a nested subquery,
-        # both Numeric values that don't have decimal places, regardless
-        # of their originating type, come back as ints with no useful
-        # typing information beyond "numeric".  So native handler
-        # must convert to int.
-        # this means our Decimal converters need to run no matter what.
-        # totally sucks.
-
-        stmt = """
-        SELECT
-            (SELECT (SELECT idata FROM foo) FROM DUAL) AS idata,
-            (SELECT CAST((SELECT ndata FROM foo) AS NUMERIC(20, 2)) FROM DUAL)
-            AS ndata,
-            (SELECT CAST((SELECT ndata2 FROM foo) AS NUMERIC(20, 2)) FROM DUAL)
-            AS ndata2,
-            (SELECT CAST((SELECT nidata FROM foo) AS NUMERIC(5, 0)) FROM DUAL)
-            AS nidata,
-            (SELECT CAST((SELECT fdata FROM foo) AS FLOAT) FROM DUAL) AS fdata
-        FROM dual
-        """
-        row = testing.db.execute(stmt).fetchall()[0]
-        eq_(
-            [type(x) for x in row],
-            [int, decimal.Decimal, int, int, decimal.Decimal]
-        )
-        eq_(
-            row,
-            (5, decimal.Decimal('45.6'), 45, 53, decimal.Decimal('45.68392'))
-        )
-
-        row = testing.db.execute(text(stmt,
-                                      typemap={
-                                          'idata': Integer(),
-                                          'ndata': Numeric(20, 2),
-                                          'ndata2': Numeric(20, 2),
-                                          'nidata': Numeric(5, 0),
-                                          'fdata': Float()})).fetchall()[0]
-        eq_(
-            [type(x) for x in row],
-            [int, decimal.Decimal, decimal.Decimal, decimal.Decimal, float]
-        )
-        eq_(
-            row,
-            (5, decimal.Decimal('45.6'), decimal.Decimal('45'),
-                decimal.Decimal('53'), 45.683920000000001)
-        )
-
-        stmt = """
-        SELECT
-                anon_1.idata AS anon_1_idata,
-                anon_1.ndata AS anon_1_ndata,
-                anon_1.ndata2 AS anon_1_ndata2,
-                anon_1.nidata AS anon_1_nidata,
-                anon_1.fdata AS anon_1_fdata
-        FROM (SELECT idata, ndata, ndata2, nidata, fdata
-        FROM (
-            SELECT
-                (SELECT (SELECT idata FROM foo) FROM DUAL) AS idata,
-                (SELECT CAST((SELECT ndata FROM foo) AS NUMERIC(20, 2))
-                FROM DUAL) AS ndata,
-                (SELECT CAST((SELECT ndata2 FROM foo) AS NUMERIC(20, 2))
-                FROM DUAL) AS ndata2,
-                (SELECT CAST((SELECT nidata FROM foo) AS NUMERIC(5, 0))
-                FROM DUAL) AS nidata,
-                (SELECT CAST((SELECT fdata FROM foo) AS FLOAT) FROM DUAL)
-                AS fdata
-            FROM dual
-        )
-        WHERE ROWNUM >= 0) anon_1
-        """
-        row = testing.db.execute(stmt).fetchall()[0]
-        eq_(
-            [type(x) for x in row],
-            [int, decimal.Decimal, int, int, decimal.Decimal]
-        )
-        eq_(
-            row,
-            (5, decimal.Decimal('45.6'), 45, 53, decimal.Decimal('45.68392'))
-        )
-
-        row = testing.db.execute(text(stmt,
-                                      typemap={
-                                          'anon_1_idata': Integer(),
-                                          'anon_1_ndata': Numeric(20, 2),
-                                          'anon_1_ndata2': Numeric(20, 2),
-                                          'anon_1_nidata': Numeric(5, 0),
-                                          'anon_1_fdata': Float()
-                                      })).fetchall()[0]
-        eq_(
-            [type(x) for x in row],
-            [int, decimal.Decimal, decimal.Decimal, decimal.Decimal, float]
-        )
-        eq_(
-            row,
-            (5, decimal.Decimal('45.6'), decimal.Decimal('45'),
-                decimal.Decimal('53'), 45.683920000000001)
-        )
-
-        row = testing.db.execute(text(
-            stmt,
-            typemap={
-                'anon_1_idata': Integer(),
-                'anon_1_ndata': Numeric(20, 2, asdecimal=False),
-                'anon_1_ndata2': Numeric(20, 2, asdecimal=False),
-                'anon_1_nidata': Numeric(5, 0, asdecimal=False),
-                'anon_1_fdata': Float(asdecimal=True)
-            })).fetchall()[0]
-        eq_(
-            [type(x) for x in row],
-            [int, float, float, float, decimal.Decimal]
-        )
-        eq_(
-            row,
-            (5, 45.6, 45, 53, decimal.Decimal('45.68392'))
-        )
-
-    @testing.provide_metadata
-    def test_reflect_dates(self):
-        metadata = self.metadata
-        Table(
-            "date_types", metadata,
-            Column('d1', sqltypes.DATE),
-            Column('d2', oracle.DATE),
-            Column('d3', TIMESTAMP),
-            Column('d4', TIMESTAMP(timezone=True)),
-            Column('d5', oracle.INTERVAL(second_precision=5)),
-        )
-        metadata.create_all()
-        m = MetaData(testing.db)
-        t1 = Table(
-            "date_types", m,
-            autoload=True)
-        assert isinstance(t1.c.d1.type, oracle.DATE)
-        assert isinstance(t1.c.d1.type, DateTime)
-        assert isinstance(t1.c.d2.type, oracle.DATE)
-        assert isinstance(t1.c.d2.type, DateTime)
-        assert isinstance(t1.c.d3.type, TIMESTAMP)
-        assert not t1.c.d3.type.timezone
-        assert isinstance(t1.c.d4.type, TIMESTAMP)
-        assert t1.c.d4.type.timezone
-        assert isinstance(t1.c.d5.type, oracle.INTERVAL)
-
-    def test_reflect_all_types_schema(self):
-        types_table = Table('all_types', MetaData(testing.db),
-                            Column('owner', String(30), primary_key=True),
-                            Column('type_name', String(30), primary_key=True),
-                            autoload=True, oracle_resolve_synonyms=True)
-        for row in types_table.select().execute().fetchall():
-            [row[k] for k in row.keys()]
-
-    @testing.provide_metadata
-    def test_raw_roundtrip(self):
-        metadata = self.metadata
-        raw_table = Table('raw', metadata,
-                          Column('id', Integer, primary_key=True),
-                          Column('data', oracle.RAW(35)))
-        metadata.create_all()
-        testing.db.execute(raw_table.insert(), id=1, data=b("ABCDEF"))
-        eq_(
-            testing.db.execute(raw_table.select()).first(),
-            (1, b("ABCDEF"))
-        )
-
-    @testing.provide_metadata
-    def test_reflect_nvarchar(self):
-        metadata = self.metadata
-        Table('tnv', metadata, Column('data', sqltypes.NVARCHAR(255)))
-        metadata.create_all()
-        m2 = MetaData(testing.db)
-        t2 = Table('tnv', m2, autoload=True)
-        assert isinstance(t2.c.data.type, sqltypes.NVARCHAR)
-
-        if testing.against('oracle+cx_oracle'):
-            # nvarchar returns unicode natively.  cx_oracle
-            # _OracleNVarChar type should be at play here.
-            assert isinstance(
-                t2.c.data.type.dialect_impl(testing.db.dialect),
-                cx_oracle._OracleNVarChar)
-
-        data = u('m’a réveillé.')
-        t2.insert().execute(data=data)
-        res = t2.select().execute().first()['data']
-        eq_(res, data)
-        assert isinstance(res, util.text_type)
-
-    @testing.provide_metadata
-    def test_char_length(self):
-        metadata = self.metadata
-        t1 = Table('t1', metadata,
-                   Column("c1", VARCHAR(50)),
-                   Column("c2", NVARCHAR(250)),
-                   Column("c3", CHAR(200)))
-        t1.create()
-        m2 = MetaData(testing.db)
-        t2 = Table('t1', m2, autoload=True)
-        eq_(t2.c.c1.type.length, 50)
-        eq_(t2.c.c2.type.length, 250)
-        eq_(t2.c.c3.type.length, 200)
-
-    @testing.provide_metadata
-    def test_long_type(self):
-        metadata = self.metadata
-
-        t = Table('t', metadata, Column('data', oracle.LONG))
-        metadata.create_all(testing.db)
-        testing.db.execute(t.insert(), data='xyz')
-        eq_(
-            testing.db.scalar(select([t.c.data])),
-            "xyz"
-        )
-
-    def test_longstring(self):
-        metadata = MetaData(testing.db)
-        testing.db.execute("""
-        CREATE TABLE Z_TEST
-        (
-          ID        NUMERIC(22) PRIMARY KEY,
-          ADD_USER  VARCHAR2(20)  NOT NULL
-        )
-        """)
-        try:
-            t = Table("z_test", metadata, autoload=True)
-            t.insert().execute(id=1.0, add_user='foobar')
-            assert t.select().execute().fetchall() == [(1, 'foobar')]
-        finally:
-            testing.db.execute("DROP TABLE Z_TEST")
-
-    @testing.fails_on('+zxjdbc', 'auto_convert_lobs not applicable')
-    def test_lobs_without_convert(self):
-        engine = testing_engine(options=dict(auto_convert_lobs=False))
-        metadata = MetaData()
-        t = Table("z_test",
-                  metadata,
-                  Column('id', Integer, primary_key=True),
-                  Column('data', Text),
-                  Column('bindata', LargeBinary))
-        t.create(engine)
-        try:
-            engine.execute(t.insert(),
-                           id=1,
-                           data='this is text',
-                           bindata=b('this is binary'))
-            row = engine.execute(t.select()).first()
-            eq_(row['data'].read(), 'this is text')
-            eq_(row['bindata'].read(), b('this is binary'))
-        finally:
-            t.drop(engine)
-
-
-class EuroNumericTest(fixtures.TestBase):
-    """
-    test the numeric output_type_handler when using non-US locale for NLS_LANG.
-    """
-
-    __only_on__ = 'oracle+cx_oracle'
-    __backend__ = True
-
-    def setup(self):
-        self.old_nls_lang = os.environ.get('NLS_LANG', False)
-        os.environ['NLS_LANG'] = "GERMAN"
-        self.engine = testing_engine()
-
-    def teardown(self):
-        if self.old_nls_lang is not False:
-            os.environ['NLS_LANG'] = self.old_nls_lang
-        else:
-            del os.environ['NLS_LANG']
-        self.engine.dispose()
-
-    def test_output_type_handler(self):
-        for stmt, exp, kw in [
-            ("SELECT 0.1 FROM DUAL", decimal.Decimal("0.1"), {}),
-            ("SELECT 15 FROM DUAL", 15, {}),
-            ("SELECT CAST(15 AS NUMERIC(3, 1)) FROM DUAL",
-             decimal.Decimal("15"), {}),
-            ("SELECT CAST(0.1 AS NUMERIC(5, 2)) FROM DUAL",
-             decimal.Decimal("0.1"), {}),
-            ("SELECT :num FROM DUAL", decimal.Decimal("2.5"),
-             {'num': decimal.Decimal("2.5")})
-        ]:
-            test_exp = self.engine.scalar(stmt, **kw)
-            eq_(
-                test_exp,
-                exp
-            )
-            assert type(test_exp) is type(exp)
-
-
-class SystemTableTablenamesTest(fixtures.TestBase):
-    __only_on__ = 'oracle'
-    __backend__ = True
-
-    def setup(self):
-        testing.db.execute("create table my_table (id integer)")
-        testing.db.execute(
-            "create global temporary table my_temp_table (id integer)"
-        )
-        testing.db.execute(
-            "create table foo_table (id integer) tablespace SYSTEM"
-        )
-
-    def teardown(self):
-        testing.db.execute("drop table my_temp_table")
-        testing.db.execute("drop table my_table")
-        testing.db.execute("drop table foo_table")
-
-    def test_table_names_no_system(self):
-        insp = inspect(testing.db)
-        eq_(
-            insp.get_table_names(), ["my_table"]
-        )
-
-    def test_temp_table_names_no_system(self):
-        insp = inspect(testing.db)
-        eq_(
-            insp.get_temp_table_names(), ["my_temp_table"]
-        )
-
-    def test_table_names_w_system(self):
-        engine = testing_engine(options={"exclude_tablespaces": ["FOO"]})
-        insp = inspect(engine)
-        eq_(
-            set(insp.get_table_names()).intersection(["my_table",
-                                                      "foo_table"]),
-            set(["my_table", "foo_table"])
-        )
-
-
-class DontReflectIOTTest(fixtures.TestBase):
-    """test that index overflow tables aren't included in
-    table_names."""
-
-    __only_on__ = 'oracle'
-    __backend__ = True
-
-    def setup(self):
-        testing.db.execute("""
-        CREATE TABLE admin_docindex(
-                token char(20),
-                doc_id NUMBER,
-                token_frequency NUMBER,
-                token_offsets VARCHAR2(2000),
-                CONSTRAINT pk_admin_docindex PRIMARY KEY (token, doc_id))
-            ORGANIZATION INDEX
-            TABLESPACE users
-            PCTTHRESHOLD 20
-            OVERFLOW TABLESPACE users
-        """)
-
-    def teardown(self):
-        testing.db.execute("drop table admin_docindex")
-
-    def test_reflect_all(self):
-        m = MetaData(testing.db)
-        m.reflect()
-        eq_(
-            set(t.name for t in m.tables.values()),
-            set(['admin_docindex'])
-        )
-
-
-class BufferedColumnTest(fixtures.TestBase, AssertsCompiledSQL):
-    __only_on__ = 'oracle'
-    __backend__ = True
-
-    @classmethod
-    def setup_class(cls):
-        global binary_table, stream, meta
-        meta = MetaData(testing.db)
-        binary_table = Table('binary_table', meta,
-                             Column('id', Integer, primary_key=True),
-                             Column('data', LargeBinary))
-        meta.create_all()
-        stream = os.path.join(
-                        os.path.dirname(__file__), "..",
-                        'binary_data_one.dat')
-        with open(stream, "rb") as file_:
-            stream = file_.read(12000)
-
-        for i in range(1, 11):
-            binary_table.insert().execute(id=i, data=stream)
-
-    @classmethod
-    def teardown_class(cls):
-        meta.drop_all()
-
-    def test_fetch(self):
-        result = binary_table.select().order_by(binary_table.c.id).\
-                                    execute().fetchall()
-        eq_(result, [(i, stream) for i in range(1, 11)])
-
-    @testing.fails_on('+zxjdbc', 'FIXME: zxjdbc should support this')
-    def test_fetch_single_arraysize(self):
-        eng = testing_engine(options={'arraysize': 1})
-        result = eng.execute(binary_table.select().
-                             order_by(binary_table.c.id)).fetchall()
-        eq_(result, [(i, stream) for i in range(1, 11)])
-
-
-class UnsupportedIndexReflectTest(fixtures.TestBase):
-    __only_on__ = 'oracle'
-    __backend__ = True
-
-    @testing.emits_warning("No column names")
-    @testing.provide_metadata
-    def test_reflect_functional_index(self):
-        metadata = self.metadata
-        Table('test_index_reflect', metadata,
-              Column('data', String(20), primary_key=True))
-        metadata.create_all()
-
-        testing.db.execute('CREATE INDEX DATA_IDX ON '
-                           'TEST_INDEX_REFLECT (UPPER(DATA))')
-        m2 = MetaData(testing.db)
-        Table('test_index_reflect', m2, autoload=True)
-
-
-def all_tables_compression_missing():
-    try:
-        testing.db.execute('SELECT compression FROM all_tables')
-        if "Enterprise Edition" not in testing.db.scalar(
-                "select * from v$version"):
-            return True
-        return False
-    except Exception:
-        return True
-
-
-def all_tables_compress_for_missing():
-    try:
-        testing.db.execute('SELECT compress_for FROM all_tables')
-        if "Enterprise Edition" not in testing.db.scalar(
-                "select * from v$version"):
-            return True
-        return False
-    except Exception:
-        return True
-
-
-class TableReflectionTest(fixtures.TestBase):
-    __only_on__ = 'oracle'
-    __backend__ = True
-
-    @testing.provide_metadata
-    @testing.fails_if(all_tables_compression_missing)
-    def test_reflect_basic_compression(self):
-        metadata = self.metadata
-
-        tbl = Table('test_compress', metadata,
-                    Column('data', Integer, primary_key=True),
-                    oracle_compress=True)
-        metadata.create_all()
-
-        m2 = MetaData(testing.db)
-
-        tbl = Table('test_compress', m2, autoload=True)
-        # Don't hardcode the exact value, but it must be non-empty
-        assert tbl.dialect_options['oracle']['compress']
-
-    @testing.provide_metadata
-    @testing.fails_if(all_tables_compress_for_missing)
-    def test_reflect_oltp_compression(self):
-        metadata = self.metadata
-
-        tbl = Table('test_compress', metadata,
-                    Column('data', Integer, primary_key=True),
-                    oracle_compress="OLTP")
-        metadata.create_all()
-
-        m2 = MetaData(testing.db)
-
-        tbl = Table('test_compress', m2, autoload=True)
-        assert tbl.dialect_options['oracle']['compress'] == "OLTP"
-
-
-class RoundTripIndexTest(fixtures.TestBase):
-    __only_on__ = 'oracle'
-    __backend__ = True
-
-    @testing.provide_metadata
-    def test_basic(self):
-        metadata = self.metadata
-
-        table = Table("sometable", metadata,
-                      Column("id_a", Unicode(255), primary_key=True),
-                      Column("id_b",
-                             Unicode(255),
-                             primary_key=True,
-                             unique=True),
-                      Column("group", Unicode(255), primary_key=True),
-                      Column("col", Unicode(255)),
-                      UniqueConstraint('col', 'group'))
-
-        # "group" is a keyword, so lower case
-        normalind = Index('tableind', table.c.id_b, table.c.group)
-        compress1 = Index('compress1', table.c.id_a, table.c.id_b,
-                          oracle_compress=True)
-        compress2 = Index('compress2', table.c.id_a, table.c.id_b, table.c.col,
-                          oracle_compress=1)
-
-        metadata.create_all()
-        mirror = MetaData(testing.db)
-        mirror.reflect()
-        metadata.drop_all()
-        mirror.create_all()
-
-        inspect = MetaData(testing.db)
-        inspect.reflect()
-
-        def obj_definition(obj):
-            return (obj.__class__,
-                    tuple([c.name for c in obj.columns]),
-                    getattr(obj, 'unique', None))
-
-        # find what the primary k constraint name should be
-        primaryconsname = testing.db.scalar(
-            text(
-                """SELECT constraint_name
-               FROM all_constraints
-               WHERE table_name = :table_name
-               AND owner = :owner
-               AND constraint_type = 'P' """),
-            table_name=table.name.upper(),
-            owner=testing.db.dialect.default_schema_name.upper())
-
-        reflectedtable = inspect.tables[table.name]
-
-        # make a dictionary of the reflected objects:
-
-        reflected = dict([(obj_definition(i), i) for i in
-                         reflectedtable.indexes
-                         | reflectedtable.constraints])
-
-        # assert we got primary key constraint and its name, Error
-        # if not in dict
-
-        assert reflected[(PrimaryKeyConstraint, ('id_a', 'id_b',
-                         'group'), None)].name.upper() \
-            == primaryconsname.upper()
-
-        # Error if not in dict
-
-        eq_(
-            reflected[(Index, ('id_b', 'group'), False)].name,
-            normalind.name
-        )
-        assert (Index, ('id_b', ), True) in reflected
-        assert (Index, ('col', 'group'), True) in reflected
-
-        idx = reflected[(Index, ('id_a', 'id_b', ), False)]
-        assert idx.dialect_options['oracle']['compress'] == 2
-
-        idx = reflected[(Index, ('id_a', 'id_b', 'col', ), False)]
-        assert idx.dialect_options['oracle']['compress'] == 1
-
-        eq_(len(reflectedtable.constraints), 1)
-        eq_(len(reflectedtable.indexes), 5)
-
-
-class SequenceTest(fixtures.TestBase, AssertsCompiledSQL):
-
-    def test_basic(self):
-        seq = Sequence('my_seq_no_schema')
-        dialect = oracle.OracleDialect()
-        assert dialect.identifier_preparer.format_sequence(seq) \
-            == 'my_seq_no_schema'
-        seq = Sequence('my_seq', schema='some_schema')
-        assert dialect.identifier_preparer.format_sequence(seq) \
-            == 'some_schema.my_seq'
-        seq = Sequence('My_Seq', schema='Some_Schema')
-        assert dialect.identifier_preparer.format_sequence(seq) \
-            == '"Some_Schema"."My_Seq"'
-
-
-class ExecuteTest(fixtures.TestBase):
-
-    __only_on__ = 'oracle'
-    __backend__ = True
-
-    def test_basic(self):
-        eq_(testing.db.execute('/*+ this is a comment */ SELECT 1 FROM '
-            'DUAL').fetchall(), [(1, )])
-
-    def test_sequences_are_integers(self):
-        seq = Sequence('foo_seq')
-        seq.create(testing.db)
-        try:
-            val = testing.db.execute(seq)
-            eq_(val, 1)
-            assert type(val) is int
-        finally:
-            seq.drop(testing.db)
-
-    @testing.provide_metadata
-    def test_limit_offset_for_update(self):
-        metadata = self.metadata
-        # oracle can't actually do the ROWNUM thing with FOR UPDATE
-        # very well.
-
-        t = Table('t1',
-                  metadata,
-                  Column('id', Integer, primary_key=True),
-                  Column('data', Integer))
-        metadata.create_all()
-
-        t.insert().execute(
-            {'id': 1, 'data': 1},
-            {'id': 2, 'data': 7},
-            {'id': 3, 'data': 12},
-            {'id': 4, 'data': 15},
-            {'id': 5, 'data': 32},
-        )
-
-        # here, we can't use ORDER BY.
-        eq_(
-            t.select(for_update=True).limit(2).execute().fetchall(),
-            [(1, 1),
-             (2, 7)]
-        )
-
-        # here, its impossible.  But we'd prefer it to raise ORA-02014
-        # instead of issuing a syntax error.
-        assert_raises_message(
-            exc.DatabaseError,
-            "ORA-02014",
-            t.select(for_update=True).limit(2).offset(3).execute
-        )
-
-
-class UnicodeSchemaTest(fixtures.TestBase):
-    __only_on__ = 'oracle'
-    __backend__ = True
-
-    @testing.provide_metadata
-    def test_quoted_column_non_unicode(self):
-        metadata = self.metadata
-        table = Table("atable", metadata,
-                      Column("_underscorecolumn",
-                             Unicode(255),
-                             primary_key=True))
-        metadata.create_all()
-
-        table.insert().execute(
-            {'_underscorecolumn': u('’é')},
-        )
-        result = testing.db.execute(
-            table.select().where(table.c._underscorecolumn == u('’é'))
-        ).scalar()
-        eq_(result, u('’é'))
-
-    @testing.provide_metadata
-    def test_quoted_column_unicode(self):
-        metadata = self.metadata
-        table = Table("atable", metadata,
-                      Column(u("méil"), Unicode(255), primary_key=True))
-        metadata.create_all()
-
-        table.insert().execute(
-            {u('méil'): u('’é')},
-        )
-        result = testing.db.execute(
-            table.select().where(table.c[u('méil')] == u('’é'))
-        ).scalar()
-        eq_(result, u('’é'))
-
-
-class DBLinkReflectionTest(fixtures.TestBase):
-    __requires__ = 'oracle_test_dblink',
-    __only_on__ = 'oracle'
-    __backend__ = True
-
-    @classmethod
-    def setup_class(cls):
-        from sqlalchemy.testing import config
-        cls.dblink = config.file_config.get('sqla_testing', 'oracle_db_link')
-
-        # note that the synonym here is still not totally functional
-        # when accessing via a different username as we do with the
-        # multiprocess test suite, so testing here is minimal
-        with testing.db.connect() as conn:
-            conn.execute("create table test_table "
-                         "(id integer primary key, data varchar2(50))")
-            conn.execute("create synonym test_table_syn "
-                         "for test_table@%s" % cls.dblink)
-
-    @classmethod
-    def teardown_class(cls):
-        with testing.db.connect() as conn:
-            conn.execute("drop synonym test_table_syn")
-            conn.execute("drop table test_table")
-
-    def test_reflection(self):
-        """test the resolution of the synonym/dblink. """
-        m = MetaData()
-
-        t = Table('test_table_syn', m, autoload=True,
-                  autoload_with=testing.db, oracle_resolve_synonyms=True)
-        eq_(list(t.c.keys()), ['id', 'data'])
-        eq_(list(t.primary_key), [t.c.id])
-
-
-class ServiceNameTest(fixtures.TestBase):
-    __only_on__ = 'oracle+cx_oracle'
-    __backend__ = True
-
-    def test_cx_oracle_service_name(self):
-        url_string = 'oracle+cx_oracle://scott:tiger@host/?service_name=hr'
-        eng = create_engine(url_string, _initialize=False)
-        cargs, cparams = eng.dialect.create_connect_args(eng.url)
-
-        assert 'SERVICE_NAME=hr' in cparams['dsn']
-        assert 'SID=hr' not in cparams['dsn']
-
-    def test_cx_oracle_service_name_bad(self):
-        url_string = 'oracle+cx_oracle://scott:tiger@host/hr1?service_name=hr2'
-        assert_raises(
-            exc.InvalidRequestError,
-            create_engine, url_string,
-            _initialize=False
-        )
-