# coding: utf-8

from sqlalchemy.testing.assertions import (
    eq_, assert_raises, assert_raises_message, AssertsExecutionResults,
    AssertsCompiledSQL)
from sqlalchemy.testing import engines, fixtures
from sqlalchemy import testing
import datetime
from sqlalchemy import (
    Table, Column, select, MetaData, text, Integer, String, Sequence, Numeric,
    DateTime, BigInteger, func, extract, SmallInteger, TypeDecorator)
from sqlalchemy import exc, schema
from sqlalchemy.dialects.postgresql import base as postgresql
import logging
import logging.handlers
from sqlalchemy.testing.mock import Mock
from sqlalchemy.engine import engine_from_config


class MiscTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):

    __only_on__ = 'postgresql'
    __backend__ = True

    @testing.provide_metadata
    def test_date_reflection(self):
        metadata = self.metadata
        Table(
            'pgdate', metadata, Column('date1', DateTime(timezone=True)),
            Column('date2', DateTime(timezone=False)))
        metadata.create_all()
        m2 = MetaData(testing.db)
        t2 = Table('pgdate', m2, autoload=True)
        assert t2.c.date1.type.timezone is True
        assert t2.c.date2.type.timezone is False

    @testing.fails_on('+zxjdbc',
                      'The JDBC driver handles the version parsing')
    def test_version_parsing(self):

        def mock_conn(res):
            return Mock(
                execute=Mock(return_value=Mock(scalar=Mock(return_value=res))))

        for string, version in [
                (
                    'PostgreSQL 8.3.8 on i686-redhat-linux-gnu, compiled by '
                    'GCC gcc (GCC) 4.1.2 20070925 (Red Hat 4.1.2-33)',
                    (8, 3, 8)),
                (
                    'PostgreSQL 8.5devel on x86_64-unknown-linux-gnu, '
                    'compiled by GCC gcc (GCC) 4.4.2, 64-bit', (8, 5)),
                (
                    'EnterpriseDB 9.1.2.2 on x86_64-unknown-linux-gnu, '
                    'compiled by gcc (GCC) 4.1.2 20080704 (Red Hat 4.1.2-50), '
                    '64-bit', (9, 1, 2)),
                (
                    '[PostgreSQL 9.2.4 ] VMware vFabric Postgres 9.2.4.0 '
                    'release build 1080137', (9, 2, 4))]:
            eq_(testing.db.dialect._get_server_version_info(mock_conn(string)),
                version)

    @testing.requires.psycopg2_compatibility
    def test_psycopg2_version(self):
        v = testing.db.dialect.psycopg2_version
        assert testing.db.dialect.dbapi.__version__.\
            startswith(".".join(str(x) for x in v))

    @testing.requires.psycopg2_compatibility
    def test_psycopg2_non_standard_err(self):
        # under pypy the name here is psycopg2cffi
        psycopg2 = testing.db.dialect.dbapi
        TransactionRollbackError = __import__(
            "%s.extensions" % psycopg2.__name__
        ).extensions.TransactionRollbackError

        exception = exc.DBAPIError.instance(
            "some statement", {}, TransactionRollbackError("foo"),
            psycopg2.Error)
        assert isinstance(exception, exc.OperationalError)

    # currently not passing with pg 9.3 that does not seem to generate
    # any notices here, would rather find a way to mock this
    @testing.requires.no_coverage
    @testing.requires.psycopg2_compatibility
    def _test_notice_logging(self):
        log = logging.getLogger('sqlalchemy.dialects.postgresql')
        buf = logging.handlers.BufferingHandler(100)
        lev = log.level
        log.addHandler(buf)
        log.setLevel(logging.INFO)
        try:
            conn = testing.db.connect()
            trans = conn.begin()
            try:
                conn.execute('create table foo (id serial primary key)')
            finally:
                trans.rollback()
        finally:
            log.removeHandler(buf)
            log.setLevel(lev)
        msgs = ' '.join(b.msg for b in buf.buffer)
        assert 'will create implicit sequence' in msgs
        assert 'will create implicit index' in msgs

    @testing.requires.psycopg2_or_pg8000_compatibility
    @engines.close_open_connections
    def test_client_encoding(self):
        c = testing.db.connect()
        current_encoding = c.execute("show client_encoding").fetchone()[0]
        c.close()

        # attempt to use an encoding that's not
        # already set
        if current_encoding == 'UTF8':
            test_encoding = 'LATIN1'
        else:
            test_encoding = 'UTF8'

        e = engines.testing_engine(options={'client_encoding': test_encoding})
        c = e.connect()
        new_encoding = c.execute("show client_encoding").fetchone()[0]
        eq_(new_encoding, test_encoding)

    @testing.requires.psycopg2_compatibility
    def test_pg_dialect_use_native_unicode_from_config(self):
        config = {
            'sqlalchemy.url': testing.db.url,
            'sqlalchemy.use_native_unicode': "false"}

        e = engine_from_config(config, _initialize=False)
        eq_(e.dialect.use_native_unicode, False)

        config = {
            'sqlalchemy.url': testing.db.url,
            'sqlalchemy.use_native_unicode': "true"}

        e = engine_from_config(config, _initialize=False)
        eq_(e.dialect.use_native_unicode, True)

    @testing.requires.psycopg2_or_pg8000_compatibility
    @engines.close_open_connections
    def test_autocommit_isolation_level(self):
        c = testing.db.connect().execution_options(
            isolation_level='AUTOCOMMIT')
        # If we're really in autocommit mode then we'll get an error saying
        # that the prepared transaction doesn't exist. Otherwise, we'd
        # get an error saying that the command can't be run within a
        # transaction.
        assert_raises_message(
            exc.ProgrammingError,
            'prepared transaction with identifier "gilberte" does not exist',
            c.execute, "commit prepared 'gilberte'")

    @testing.fails_on('+zxjdbc',
                      "Can't infer the SQL type to use for an instance "
                      "of org.python.core.PyObjectDerived.")
    def test_extract(self):
        fivedaysago = datetime.datetime.now() \
            - datetime.timedelta(days=5)
        for field, exp in ('year', fivedaysago.year), \
                ('month', fivedaysago.month), ('day', fivedaysago.day):
            r = testing.db.execute(
                select([
                    extract(field, func.now() + datetime.timedelta(days=-5))])
            ).scalar()
            eq_(r, exp)

    def test_checksfor_sequence(self):
        meta1 = MetaData(testing.db)
        seq = Sequence('fooseq')
        t = Table(
            'mytable', meta1,
            Column('col1', Integer, seq)
        )
        seq.drop()
        try:
            testing.db.execute('CREATE SEQUENCE fooseq')
            t.create(checkfirst=True)
        finally:
            t.drop(checkfirst=True)

    def test_schema_roundtrips(self):
        meta = MetaData(testing.db)
        users = Table(
            'users', meta, Column(
                'id', Integer, primary_key=True), Column(
                'name', String(50)), schema='test_schema')
        users.create()
        try:
            users.insert().execute(id=1, name='name1')
            users.insert().execute(id=2, name='name2')
            users.insert().execute(id=3, name='name3')
            users.insert().execute(id=4, name='name4')
            eq_(users.select().where(users.c.name == 'name2')
                .execute().fetchall(), [(2, 'name2')])
            eq_(users.select(use_labels=True).where(
                users.c.name == 'name2').execute().fetchall(), [(2, 'name2')])
            users.delete().where(users.c.id == 3).execute()
            eq_(users.select().where(users.c.name == 'name3')
                .execute().fetchall(), [])
            users.update().where(users.c.name == 'name4'
                                 ).execute(name='newname')
            eq_(users.select(use_labels=True).where(
                users.c.id == 4).execute().fetchall(), [(4, 'newname')])
        finally:
            users.drop()

    def test_preexecute_passivedefault(self):
        """test that when we get a primary key column back from
        reflecting a table which has a default value on it, we pre-
        execute that DefaultClause upon insert."""

        try:
            meta = MetaData(testing.db)
            testing.db.execute("""
             CREATE TABLE speedy_users
             (
                 speedy_user_id   SERIAL     PRIMARY KEY,

                 user_name        VARCHAR    NOT NULL,
                 user_password    VARCHAR    NOT NULL
             );
            """)
            t = Table('speedy_users', meta, autoload=True)
            r = t.insert().execute(user_name='user',
                                   user_password='lala')
            assert r.inserted_primary_key == [1]
            l = t.select().execute().fetchall()
            assert l == [(1, 'user', 'lala')]
        finally:
            testing.db.execute('drop table speedy_users')

    @testing.fails_on('+zxjdbc', 'psycopg2/pg8000 specific assertion')
    @testing.requires.psycopg2_or_pg8000_compatibility
    def test_numeric_raise(self):
        stmt = text(
            "select cast('hi' as char) as hi", typemap={'hi': Numeric})
        assert_raises(exc.InvalidRequestError, testing.db.execute, stmt)

    @testing.only_if(
        "postgresql >= 8.2", "requires standard_conforming_strings")
    def test_serial_integer(self):

        class BITD(TypeDecorator):
            impl = Integer

            def load_dialect_impl(self, dialect):
                if dialect.name == 'postgresql':
                    return BigInteger()
                else:
                    return Integer()

        for version, type_, expected in [
            (None, Integer, 'SERIAL'),
            (None, BigInteger, 'BIGSERIAL'),
            ((9, 1), SmallInteger, 'SMALLINT'),
            ((9, 2), SmallInteger, 'SMALLSERIAL'),
            (None, postgresql.INTEGER, 'SERIAL'),
            (None, postgresql.BIGINT, 'BIGSERIAL'),
            (
                None, Integer().with_variant(BigInteger(), 'postgresql'),
                'BIGSERIAL'),
            (
                None, Integer().with_variant(postgresql.BIGINT, 'postgresql'),
                'BIGSERIAL'),
            (
                (9, 2), Integer().with_variant(SmallInteger, 'postgresql'),
                'SMALLSERIAL'),
            (None, BITD(), 'BIGSERIAL')
        ]:
            m = MetaData()

            t = Table('t', m, Column('c', type_, primary_key=True))

            if version:
                dialect = postgresql.dialect()
                dialect._get_server_version_info = Mock(return_value=version)
                dialect.initialize(testing.db.connect())
            else:
                dialect = testing.db.dialect

            ddl_compiler = dialect.ddl_compiler(dialect, schema.CreateTable(t))
            eq_(
                ddl_compiler.get_column_specification(t.c.c),
                "c %s NOT NULL" % expected
            )