# coding: utf-8

from test.lib.testing import eq_
from sqlalchemy import *
from sqlalchemy import types as sqltypes, exc
from sqlalchemy.sql import table, column
from test.lib import *
from test.lib.testing import eq_, assert_raises, assert_raises_message
from test.lib.engines import testing_engine
from sqlalchemy.dialects.oracle import cx_oracle, base as oracle
from sqlalchemy.engine import default
from sqlalchemy.util import jython
from sqlalchemy.util.compat import decimal
import datetime
import os


class OutParamTest(fixtures.TestBase, AssertsExecutionResults):
    __only_on__ = 'oracle+cx_oracle'

    @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 CompileTest(fixtures.TestBase, AssertsCompiledSQL):

    __dialect__ = oracle.OracleDialect()

    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_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.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 <= '
                            ':ROWNUM_1) WHERE ora_rn > :ora_rn_1')

        c = s.compile(dialect=oracle.OracleDialect())
        assert t.c.col1 in set(c.result_map['col1'][1])
        s = select([s.c.col1, s.c.col2])
        self.assert_compile(s,
                            '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 <= :ROWNUM_1) WHERE ora_rn > '
                            ':ora_rn_1)')

        self.assert_compile(s,
                            '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 <= :ROWNUM_1) WHERE ora_rn > '
                            ':ora_rn_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 <= '
                            ':ROWNUM_1) WH
ERE ora_rn > :ora_rn_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 <= :ROWNUM_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 <= '
                            ':ROWNUM_1) WHERE ora_rn > :ora_rn_1 FOR '
                            'UPDATE')

    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 "
                "<= :ROWNUM_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 > :ora_rn_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 <= :ROWNUM_1) WHERE ora_rn > "
                ":ora_rn_1",
                dialect=dialect)

    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',
      