<commit_before># vim: set fileencoding=utf8 :
from __future__ import with_statement
from six import StringIO
import unittest
import sys
from decimal import Decimal, getcontext
import logging
from time import sleep
from datetime import datetime, date, time
import uuid
from dateutil.tz import tzoffset, tzutc
from six import text_type
from six.moves import xrange
import binascii
from pytds import (
    connect, ProgrammingError, TimeoutError, Time, SimpleLoadBalancer, LoginError,
    Error, IntegrityError, Timestamp, DataError, DECIMAL, Date, Binary, DateTime,
    IS_TDS73_PLUS, IS_TDS71_PLUS, NotSupportedError, TDS73, TDS71, TDS72,
    output, default, InterfaceError, TDS_ENCRYPTION_OFF)
from pytds.tds import (
    _TdsSocket, _TdsSession, TDS_ENCRYPTION_REQUIRE, Column, Bit, Int, SmallInt,
    NVarChar72, TinyInt, IntN, BigInt, Real, Float, FloatN, Collation,
    )
from pytds.dbapi import _TdsLogin
from . import dbapi20
import pytds

# set decimal precision to match mssql maximum precision
getcontext().prec = 38

try:
    from . import settings
except:
    print('Settings module is not found, please create settings module and specify HOST, DATATABSE, USER and PASSWORD there')
    sys.exit(1)

#logging.basicConfig(level='DEBUG')
#logging.basicConfig(level='INFO')
logging.basicConfig()


class TestCase(unittest.TestCase):
    def setUp(self):
        kwargs = settings.CONNECT_KWARGS.copy()
        kwargs['database'] = 'master'
        self.conn = connect(*settings.CONNECT_ARGS, **kwargs)

    def tearDown(self):
        self.conn.close()


class DbTestCase(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        kwargs = settings.CONNECT_KWARGS.copy()
        kwargs['database'] = 'master'
        kwargs['autocommit'] = True
        with connect(**kwargs) as conn:
            with conn.cursor() as cur:
                try:
                    cur.execute('drop database [{}]'.format(settings.DATABASE))
                except:
                    pass
                cur.execute('create database [{}]'.format(settings.DATABASE))

    @classmethod
    def tearDownClass(cls):
        kwargs = settings.CONNECT_KWARGS.copy()
        kwargs['database'] = 'master'
        kwargs['autocommit'] = True
        with connect(**kwargs) as conn:
            with conn.cursor() as cur:
                cur.execute('drop database [{}]'.format(settings.DATABASE))

    def setUp(self):
        self.conn = pytds.connect(*settings.CONNECT_ARGS, **settings.CONNECT_KWARGS)

    def tearDown(self):
        self.conn.close()


class TestCase2(TestCase):
    def test_all(self):
        cur = self.conn.cursor()
        with self.assertRaises(ProgrammingError):
            cur.execute(u'select ')
        assert 'abc' == cur.execute_scalar("select cast('abc' as nvarchar(max)) as fieldname")
        assert b'abc' == cur.execute_scalar("select cast('abc' as varbinary(max)) as fieldname")
        #assert 12 == cur.execute_scalar('select cast(12 as bigint) as fieldname')
        assert 12 == cur.execute_scalar('select cast(12 as smallint) as fieldname')
        assert -12 == cur.execute_scalar('select -12 as fieldname')
        assert 12 == cur.execute_scalar('select cast(12 as tinyint) as fieldname')
        assert True == cur.execute_scalar('select cast(1 as bit) as fieldname')
        assert 5.1 == cur.execute_scalar('select cast(5.1 as float) as fieldname')
        cur.execute("select 'test', 20")
        assert ('test', 20) == cur.fetchone()
        assert 'test' == cur.execute_scalar("select 'test' as fieldname")
        assert 'test' == cur.execute_scalar("select N'test' as fieldname")
        assert 'test' == cur.execute_scalar("select cast(N'test' as ntext) as fieldname")
        assert 'test' == cur.execute_scalar("select cast(N'test' as text) as fieldname")
        self.assertEqual('test ', cur.execute_scalar("select cast(N'test' as char(5)) as fieldname"))
        self.assertEqual('test ', cur.execute_scalar("select cast(N'test' as nchar(5)) as fieldname"))
        assert b'test' == cur.execute_scalar("select cast('test' as varbinary(4)) as fieldname")
        assert b'test' == cur.execute_scalar("select cast('test' as image) as fieldname")
        assert None == cur.execute_scalar("select cast(NULL as image) as fieldname")
        assert None == cur.execute_scalar("select cast(NULL as varbinary(10)) as fieldname")
        assert None == cur.execute_scalar("select cast(NULL as ntext) as fieldname")
        assert None == cur.execute_scalar("select cast(NULL as nvarchar(max)) as fieldname")
        assert None == cur.execute_scalar("select cast(NULL as xml)")
        self.assertEqual(None, cur.execute_scalar("select cast(NULL as varchar(max)) as fieldname"))
        assert None == cur.execute_scalar("select cast(NULL as nvarchar(10)) as fieldname")
        assert None == cur.execute_scalar("select cast(NULL as varchar(10)) as fieldname")
        assert None == cur.execute_scalar("select cast(NULL as nchar(10)) as fieldname")
        assert None == cur.execute_scalar("select cast(NULL as char(10)) as fieldname")
        assert None == cur.execute_scalar("select cast(NULL as char(10)) as fieldname")
        self.assertEqual(u'Iñtërnâtiônàlizætiøn1', cur.execute_scalar('select %s', (u'Iñtërnâtiônàlizætiøn1'.encode('utf8'),)))
        assert 5 == cur.execute_scalar('select 5 as fieldname')

    def test_decimals(self):
        cur = self.conn.cursor()
        assert Decimal(12) == cur.execute_scalar('select cast(12 as decimal) as fieldname')
        assert Decimal(-12) == cur.execute_scalar('select cast(-12 as decimal) as fieldname')
        assert Decimal('123456.12345') == cur.execute_scalar("select cast('123456.12345'as decimal(20,5)) as fieldname")
        assert Decimal('-123456.12345') == cur.execute_scalar("select cast('-123456.12345'as decimal(20,5)) as fieldname")

    def test_money(self):
        cur = self.conn.cursor()
        assert Decimal('0') == cur.execute_scalar("select cast('0' as money) as fieldname")
        assert Decimal('1') == cur.execute_scalar("select cast('1' as money) as fieldname")
        self.assertEqual(Decimal('1.5555'), cur.execute_scalar("select cast('1.5555' as money) as fieldname"))
        assert Decimal('1234567.5555') == cur.execute_scalar("select cast('1234567.5555' as money) as fieldname")
        assert Decimal('-1234567.5555') == cur.execute_scalar("select cast('-1234567.5555' as money) as fieldname")
        assert Decimal('12345.55') == cur.execute_scalar("select cast('12345.55' as smallmoney) as fieldname")

    def test_timeout(self):
        kwargs = settings.CONNECT_KWARGS.copy()
        kwargs['database'] = 'master'
        kwargs['login_timeout'] = 1
        with connect(*settings.CONNECT_ARGS, **kwargs) as conn:
            cur = conn.cursor()
            with self.assertRaises(TimeoutError):
                cur.execute("waitfor delay '00:00:05'")
            cur.execute('select 1')

    def test_strs(self):
        cur = self.conn.cursor()
        self.assertIsInstance(cur.execute_scalar("select 'test'"), text_type)

    #def test_mars_sessions_recycle_ids(self):
    #    if not self.conn.mars_enabled:
    #        self.skipTest('Only relevant to mars')
    #    for _ in xrange(2 ** 16 + 1):
    #        cur = self.conn.cursor()
    #        cur.close()

    def test_smp(self):
        if not self.conn.mars_enabled:
            self.skipTest('Only relevant to mars')
        sess = self.conn._conn._smp_manager.create_session()
        self.assertEqual(sess._state, 'SESSION ESTABLISHED')
        sess.close()
        self.assertEqual(sess._state, 'CLOSED')

    def test_cursor_env(self):
        with self.conn.cursor() as cur:
            cur.execute('use master')
            self.assertEqual(cur.execute_scalar('select DB_NAME()'), 'master')

    def test_empty_query(self):
        with self.conn.cursor() as cur:
            cur.execute('')
            self.assertIs(None, cur.description)

    def test_parameters_ll(self):
        def test_val(typ, val):
            with self.conn.cursor() as cur:
                param = Column(type=typ, value=val)
                cur.execute('select %s', [param])
                self.assertTupleEqual(cur.fetchone(), (val,))
                self.assertIs(cur.fetchone(), None)

        test_val(self.conn._conn.BitN, True)
        test_val(self.conn._conn.BitN, False)
        test_val(self.conn._conn.BitN, None)
        test_val(self.conn._conn.IntN(1), 255)
        test_val(self.conn._conn.IntN(2), 2 ** 15 - 1)
        test_val(self.conn._conn.IntN(4), 2 ** 31 - 1)
        test_val(self.conn._conn.IntN(8), 2 ** 63 - 1)
        test_val(self.conn._conn.IntN(4), None)
        #test_val(self.conn._conn.Real, 0.25)
        #test_val(self.conn._conn.Float, 0.25)
        test_val(self.conn._conn.FloatN(4), 0.25)
        test_val(self.conn._conn.FloatN(8), 0.25)
        test_val(self.conn._conn.FloatN(4), None)
        #test_val(self.conn._conn.SmallDateTime, datetime(1900, 1, 1, 0, 0, 0))
        #test_val(self.conn._conn.SmallDateTime, datetime(2079, 6, 6, 23, 59, 0))
        #test_val(self.conn._conn.DateTime, datetime(1753, 1, 1, 0, 0, 0))
        #test_val(self.conn._conn.DateTime, datetime(9999, 12, 31, 23, 59, 59, 990000))
        test_val(self.conn._conn.DateTimeN(4), datetime(1900, 1, 1, 0, 0, 0))
        test_val(self.conn._conn.DateTimeN(4), datetime(2079, 6, 6, 23, 59, 0))
        test_val(self.conn._conn.DateTimeN(8), datetime(1753, 1, 1, 0, 0, 0))
        test_val(self.conn._conn.DateTimeN(8), datetime(9999, 12, 31, 23, 59, 59, 990000))
        test_val(self.conn._conn.DateTimeN(8), None)
        test_val(self.conn._conn.Date, date(1, 1, 1))
        test_val(self.conn._conn.Date, date(9999, 12, 31))
        test_val(self.conn._conn.Date, None)
        test_val(self.conn._conn.Time(0), time(0, 0, 0))
        test_val(self.conn._conn.Time(6), time(23, 59, 59, 999999))
        test_val(self.conn._conn.Time(0), None)
        test_val(self.conn._conn.DateTime2(0), datetime(1, 1, 1, 0, 0, 0))
        test_val(self.conn._conn.DateTime2(6), datetime(9999, 12, 31, 23, 59, 59, 999999))
        test_val(self.conn._conn.DateTime2(0), None)
        test_val(self.conn._conn.DateTimeOffset(6), datetime(9999, 12, 31, 23, 59, 59, 999999, tzutc()))
        test_val(self.conn._conn.DateTimeOffset(6), datetime(9999, 12, 31, 23, 59, 59, 999999, tzoffset('', 14 * 60)))
        test_val(self.conn._conn.DateTimeOffset(0), datetime(1, 1, 1, 0, 0, 0, tzinfo=tzoffset('', -14 * 60)))
        #test_val(self.conn._conn.DateTimeOffset(0), datetime(1, 1, 1, 0, 0, 0, tzinfo=tzoffset('', 14 * 60)))
        test_val(self.conn._conn.DateTimeOffset(6), None)
        test_val(self.conn._conn.Decimal(6, 38), Decimal('123.456789'))
        test_val(self.conn._conn.Decimal(6, 38), None)
        #test_val(self.conn._conn.SmallMoney, Decimal('214748.3647'))
        #test_val(self.conn._conn.SmallMoney, Decimal('-214748.3648'))
        #test_val(self.conn._conn.Money, Decimal('922337203685477.5807'))
        #test_val(self.conn._conn.Money, Decimal('-922337203685477.5808'))
        test_val(self.conn._conn.MoneyN(4), Decimal('-214748.3648'))
        test_val(self.conn._conn.MoneyN(4), Decimal('214748.3647'))
        test_val(self.conn._conn.MoneyN(8), Decimal('922337203685477.5807'))
        test_val(self.conn._conn.MoneyN(8), Decimal('-922337203685477.5808'))
        test_val(self.conn._conn.MoneyN(8), None)
        test_val(self.conn._conn.UniqueIdentifier, None)
        test_val(self.conn._conn.UniqueIdentifier, uuid.uuid4())
        test_val(self.conn._conn.SqlVariant(10), None)
        #test_val(self.conn._conn.SqlVariant(10), 100)
        test_val(self.conn._conn.VarBinary(10), b'')
        test_val(self.conn._conn.VarBinary(10), b'testtest12')
        test_val(self.conn._conn.VarBinary(10), None)
        test_val(self.conn._conn.VarBinary(8000), b'x' * 8000)
        test_val(self.conn._conn.long_binary_type(), None)
        test_val(self.conn._conn.long_binary_type(), b'')
        test_val(self.conn._conn.long_binary_type(), b'testtest12')
        test_val(self.conn._conn.long_binary_type(), b'x' * (10 ** 6))
        test_val(self.conn._conn.NVarChar(10), u'')
        test_val(self.conn._conn.NVarChar(10), u'testtest12')
        test_val(self.conn._conn.NVarChar(10), None)
        test_val(self.conn._conn.NVarChar(4000), u'x' * 4000)
        test_val(self.conn._conn.long_string_type(), None)
        test_val(self.conn._conn.long_string_type(), 'test')
        test_val(self.conn._conn.long_string_type(), 'x' * (10 ** 6))
        #test_val(self.conn._conn.Xml(), '<root></root>')
        #test_val(self.conn._conn.Image(10, []), None)
        #test_val(self.conn._conn.Image(10, ['']), None)
        #test_val(self.conn._conn.Image(10, ['']), b'test')

    def test_parameters(self):
        def test_val(val):
            with self.conn.cursor() as cur:
                cur.execute('select %s', (val,))
                self.assertTupleEqual(cur.fetchone(), (val,))
                self.assertIs(cur.fetchone(), None)

        test_val(u'hello')
        test_val(123)
        test_val(-123)
        test_val(123.12)
        test_val(-123.12)
        test_val(10 ** 20)
        test_val(10 ** 38 - 1)
        test_val(-10 ** 38 + 1)
        test_val(datetime(2011, 2, 3, 10, 11, 12, 3000))
        test_val(Decimal('1234.567'))
        test_val(Decimal('1234000'))
        test_val(Decimal('9' * 38))
        test_val(Decimal('0.' + '9' * 38))
        test_val(-Decimal('9' * 38))
        test_val(Decimal('1E10'))
        test_val(Decimal('1E-10'))
        test_val(Decimal('0.{0}1'.format('0' * 37)))
        test_val(None)
        test_val('hello')
        test_val('')
        test_val('x' * 5000)
        test_val(Binary(b'\x00\x01\x02'))
        test_val(Binary(b'x' * 5000))
        test_val(2 ** 63 - 1)
        test_val(False)
        test_val(True)
        test_val(uuid.uuid4())

    def test_overlimit(self):
        def test_val(val):
            with self.conn.cursor() as cur:
                cur.execute('select %s', (val,))
                self.assertTupleEqual(cur.fetchone(), (val,))
                self.assertIs(cur.fetchone(), None)

        with self.assertRaises(DataError):
            test_val(-10 ** 38)
        ##cur.execute('select %s', '\x00'*(2**31))
        with self.assertRaises(DataError):
            test_val(Decimal('1' + '0' * 38))
        with self.assertRaises(DataError):
            test_val(Decimal('-1' + '0' * 38))
        with self.assertRaises(DataError):
            test_val(Decimal('1E38'))

    def test_description(self):
        with self.conn.cursor() as cur:
            cur.execute('select cast(12.65 as decimal(4,2)) as testname')
            self.assertEqual(cur.description[0][0], 'testname')
            self.assertEqual(cur.description[0][1], DECIMAL)
            self.assertEqual(cur.description[0][4], 4)
            self.assertEqual(cur.description[0][5], 2)

    def test_bug4(self):
        with self.conn.cursor() as cur:
            cur.execute('''
            set transaction isolation level read committed
            select 1
            ''')
            self.assertEqual(cur.fetchall(), [(1,)])

    def test_bad_collation(self):
        with self.conn.cursor() as cur:
            try:
                cur.execute_scalar('select cast(0x90 as varchar)')
            except:
                pass
            self.assertEqual(1, cur.execute_scalar('select 1'))

    def test_get_instances(self):
        if not hasattr(settings, 'BROWSER_ADDRESS'):
            return unittest.skip('BROWSER_ADDRESS setting is not defined')
        pytds.tds.tds7_get_instances(settings.BROWSER_ADDRESS)


class DbTests(DbTestCase):
    def test_autocommit(self):
        self.assertFalse(self.conn.autocommit)
        with self.conn.cursor() as cur:
            cur.execute('create table test_autocommit(field int)')
            self.conn.commit()
            self.assertEqual(self.conn._trancount(), 1)
            cur.execute('insert into test_autocommit(field) values(1)')
            self.assertEqual(self.conn._trancount(), 1)
            cur.execute('select field from test_autocommit')
            row = cur.fetchone()
            self.conn.rollback()
            cur.execute('select field from test_autocommit')
            row = cur.fetchone()
            self.assertFalse(row)

            self.conn.autocommit = True
            cur.execute('insert into test_autocommit(field) values(1)')
            self.assertEqual(self.conn._trancount(), 0)

    def _test_bulk_type(self, typ, value):
        with self.conn.cursor() as cur:
            cur.execute('create table bulk_insert_table_ll(c1 {})'.format(typ.get_declaration()))
            cur._session.submit_plain_query('insert bulk bulk_insert_table_ll (c1 {})'.format(typ.get_declaration()))
            cur._session.process_simple_request()
            col1 = Column('c1', typ, flags=Column.fNullable)
            metadata = [col1]
            cur._session.submit_bulk(metadata, [(value,)])
            cur._session.process_simple_request()
        with self.conn.cursor() as cur:
            cur.execute('select c1 from bulk_insert_table_ll')
            self.assertTupleEqual(cur.fetchone(), (value,))
            self.assertIs(cur.fetchone(), None)
            cur.execute('drop table bulk_insert_table_ll')

    def test_bulk_insert_low_level(self):
        self._test_bulk_type(Bit.instance, True)
        self._test_bulk_type(Bit.instance, False)
        self._test_bulk_type(Int.instance, 2 ** 31 - 1)
        self._test_bulk_type(Int.instance, -2 ** 31)
        self._test_bulk_type(SmallInt.instance, -2 ** 15)
        self._test_bulk_type(SmallInt.instance, 2 ** 15 - 1)
        self._test_bulk_type(TinyInt.instance, 255)
        self._test_bulk_type(TinyInt.instance, 0)
        self._test_bulk_type(BigInt.instance, 2 ** 63 - 1)
        self._test_bulk_type(BigInt.instance, -2 ** 63)
        self._test_bulk_type(IntN(1), 255)
        self._test_bulk_type(IntN(2), 2 ** 15 - 1)
        self._test_bulk_type(IntN(4), 2 ** 31 - 1)
        self._test_bulk_type(IntN(8), 2 ** 63 - 1)
        self._test_bulk_type(IntN(4), None)
        self._test_bulk_type(Real.instance, 0.25)
        self._test_bulk_type(Float.instance, 0.25)
        self._test_bulk_type(FloatN(4), 0.25)
        self._test_bulk_type(FloatN(8), 0.25)
        self._test_bulk_type(FloatN(4), None)
        self._test_bulk_type(self.conn._conn.NVarChar(10), u'')
        self._test_bulk_type(self.conn._conn.NVarChar(10), u'testtest12')
        self._test_bulk_type(self.conn._conn.NVarChar(10), None)
        self._test_bulk_type(self.conn._conn.NVarChar(4000), u'x' * 4000)
        self._test_bulk_type(self.conn._conn.VarBinary(10), b'')
        self._test_bulk_type(self.conn._conn.VarBinary(10), b'testtest12')
        self._test_bulk_type(self.conn._conn.VarBinary(10), None)
        self._test_bulk_type(self.conn._conn.VarBinary(8000), b'x' * 8000)
        self._test_bulk_type(self.conn._conn.SmallDateTime, datetime(1900, 1, 1, 0, 0, 0))
        self._test_bulk_type(self.conn._conn.SmallDateTime, datetime(2079, 6, 6, 23, 59, 0))
        self._test_bulk_type(self.conn._conn.DateTime, datetime(1753, 1, 1, 0, 0, 0))
        self._test_bulk_type(self.conn._conn.DateTime, datetime(9999, 12, 31, 23, 59, 59, 990000))
        self._test_bulk_type(self.conn._conn.DateTimeN(4), datetime(1900, 1, 1, 0, 0, 0))
        self._test_bulk_type(self.conn._conn.DateTimeN(8), datetime(9999, 12, 31, 23, 59, 59, 990000))
        self._test_bulk_type(self.conn._conn.DateTimeN(8), None)
        self._test_bulk_type(self.conn._conn.Date, date(1, 1, 1))
        self._test_bulk_type(self.conn._conn.Date, date(9999, 12, 31))
        self._test_bulk_type(self.conn._conn.Date, None)
        self._test_bulk_type(self.conn._conn.Time(0), time(0, 0, 0))
        self._test_bulk_type(self.conn._conn.Time(6), time(23, 59, 59, 999999))
        self._test_bulk_type(self.conn._conn.Time(0), None)
        self._test_bulk_type(self.conn._conn.DateTime2(0), datetime(1, 1, 1, 0, 0, 0))
        self._test_bulk_type(self.conn._conn.DateTime2(6), datetime(9999, 12, 31, 23, 59, 59, 999999))
        self._test_bulk_type(self.conn._conn.DateTime2(0), None)
        self._test_bulk_type(self.conn._conn.DateTimeOffset(6), datetime(9999, 12, 31, 23, 59, 59, 999999, tzutc()))
        self._test_bulk_type(self.conn._conn.DateTimeOffset(6), datetime(9999, 12, 31, 23, 59, 59, 999999, tzoffset('', 14 * 60)))
        self._test_bulk_type(self.conn._conn.DateTimeOffset(0), datetime(1, 1, 1, 0, 0, 0, tzinfo=tzoffset('', -14 * 60)))
        #self._test_bulk_type(self.conn._conn.DateTimeOffset(0), datetime(1, 1, 1, 0, 0, 0, tzinfo=tzoffset('', 14 * 60)))
        self._test_bulk_type(self.conn._conn.DateTimeOffset(6), None)
        self._test_bulk_type(self.conn._conn.Decimal(6, 38), Decimal('123.456789'))
        self._test_bulk_type(self.conn._conn.Decimal(6, 38), None)
        self._test_bulk_type(self.conn._conn.SmallMoney, Decimal('214748.3647'))
        self._test_bulk_type(self.conn._conn.SmallMoney, Decimal('-214748.3648'))
        self._test_bulk_type(self.conn._conn.Money, Decimal('922337203685477.5807'))
        self._test_bulk_type(self.conn._conn.Money, Decimal('-922337203685477.5808'))
        self._test_bulk_type(self.conn._conn.MoneyN(4), Decimal('214748.3647'))
        self._test_bulk_type(self.conn._conn.MoneyN(8), Decimal('922337203685477.5807'))
        self._test_bulk_type(self.conn._conn.MoneyN(8), None)
        self._test_bulk_type(self.conn._conn.UniqueIdentifier, None)
        self._test_bulk_type(self.conn._conn.UniqueIdentifier, uuid.uuid4())
        self._test_bulk_type(self.conn._conn.SqlVariant(10), None)
        #self._test_bulk_type(self.conn._conn.SqlVariant(10), 100)
        self._test_bulk_type(self.conn._conn.long_binary_type(), None)
        #self._test_bulk_type(self.conn._conn.long_binary_type(), b'')
        #self._test_bulk_type(self.conn._conn.long_binary_type(), b'testtest12')
        self._test_bulk_type(self.conn._conn.long_string_type(), None)
        #self._test_bulk_type(self.conn._conn.long_string_type(), 'test')
        #self._test_bulk_type(self.conn._conn.Image(10, []), None)
        #self._test_bulk_type(self.conn._conn.Image(10, ['']), None)
        #self._test_bulk_type(self.conn._conn.Image(10, ['']), b'test')

    #def test_bulk_insert(self):
    #    with self.conn.cursor() as cur:
    #        cur.execute('create table bulk_insert_table(num int, data varchar(100))')
    #        f = StringIO("42\tfoo\n74\tbar\n")
    #        cur.copy_to(f, 'bulk_insert_table', columns=('num', 'data'))
    #        cur.execute('select num, data from bulk_insert_table')
    #        self.assertListEqual(cur.fetchall(), [(42, 'foo'), (74, 'bar')])

    def test_table_selects(self):
        cur = self.conn.cursor()
        cur.execute(u'''
        create table #testtable (id int, _text text, _xml xml, vcm varchar(max), vc varchar(10))
        ''')
        cur.execute(u'''
        insert into #testtable (id, _text, _xml, vcm, vc) values (1, 'text', '<root/>', '', NULL)
        ''')
        cur.execute('select id from #testtable order by id')
        self.assertEqual([(1,)], cur.fetchall())

        cur = self.conn.cursor()
        cur.execute('select _text from #testtable order by id')
        self.assertEqual([(u'text',)], cur.fetchall())

        cur = self.conn.cursor()
        cur.execute('select _xml from #testtable order by id')
        self.assertEqual([('<root/>',)], cur.fetchall())

        cur = self.conn.cursor()
        cur.execute('select id, _text, _xml, vcm, vc from #testtable order by id')
        self.assertTupleEqual((1, 'text', '<root/>', '', None), cur.fetchone())

        cur = self.conn.cursor()
        cur.execute('select vc from #testtable order by id')
        self.assertEqual([(None,)], cur.fetchall())

        cur = self.conn.cursor()
        cur.execute('insert into #testtable (_xml) values (%s)', ('<some/>',))

        cur = self.conn.cursor()
        cur.execute(u'drop table #testtable')

    def test_stored_proc(self):
        cur = self.conn.cursor()
        cur.execute('''
        create procedure testproc (@param int, @add int = 2, @outparam int output)
        as
        begin
            set nocount on
            --select @param
            set @outparam = @param + @add
            return @outparam
        end
        ''')
        val = 45
        #params = {'@param': val, '@outparam': output(None), '@add': 1}
        values = cur.callproc('testproc', (val, default, output(None, 1)))
        #self.assertEqual(cur.fetchall(), [(val,)])
        self.assertEqual(val + 2, values[2])
        self.assertEqual(val + 2, cur.get_proc_return_status())

    def test_fetchone(self):
        with self.conn.cursor() as cur:
            cur.execute('select 10; select 12')
            self.assertEqual((10,), cur.fetchone())
            self.assertTrue(cur.nextset())
            self.assertEqual((12,), cur.fetchone())
            self.assertFalse(cur.nextset())

    def test_fetchall(self):
        with self.conn.cursor() as cur:
            cur.execute('select 10; select 12')
            self.assertEqual([(10,)], cur.fetchall())
            self.assertTrue(cur.nextset())
            self.assertEqual([(12,)], cur.fetchall())
            self.assertFalse(cur.nextset())

    def test_cursor_closing(self):
        with self.conn.cursor() as cur:
            cur.execute('select 10; select 12')
            cur.fetchone()
        with self.conn.cursor() as cur2:
            cur2.execute('select 20')
            cur2.fetchone()

    def test_transactions(self):
        self.conn.autocommit = False
        with self.conn.cursor() as cur:
            cur.execute('''
            create table testtable (field datetime)
            ''')
            cur.execute("select object_id('testtable')")
            self.assertNotEqual((None,), cur.fetchone())

            self.assertEqual(1, self.conn._trancount())

            self.conn.rollback()

            self.assertEqual(1, self.conn._trancount())

            cur.execute("select object_id('testtable')")
            self.assertEqual((None,), cur.fetchone())

            cur.execute('''
            create table testtable (field datetime)
            ''')

            self.conn.commit()

            cur.execute("select object_id('testtable')")
            self.assertNotEqual((None,), cur.fetchone())

        with self.conn.cursor() as cur:
            cur.execute('''
            if object_id('testtable') is not null
                drop table testtable
            ''')
        self.conn.commit()

    def test_multi_packet(self):
        cur = self.conn.cursor()
        param = 'x' * (self.conn._conn.main_session._writer.bufsize * 3)
        cur.execute('select %s', (param,))
        self.assertEqual([(param, )], cur.fetchall())

    def test_big_request(self):
        with self.conn.cursor() as cur:
            param = 'x' * 5000
            params = (10, datetime(2012, 11, 19, 1, 21, 37, 3000), param, 'test')
            cur.execute('select %s, %s, %s, %s', params)
            self.assertEqual([params], cur.fetchall())

    def test_row_count(self):
        cur = self.conn.cursor()
        cur.execute('''
        create table testtable (field int)
        ''')
        cur.execute('insert into testtable (field) values (1)')
        self.assertEqual(cur.rowcount, 1)
        cur.execute('insert into testtable (field) values (2)')
        self.assertEqual(cur.rowcount, 1)
        cur.execute('select * from testtable')
        cur.fetchall()
        self.assertEqual(cur.rowcount, 2)

    def test_no_rows(self):
        cur = self.conn.cursor()
        cur.execute('''
        create table testtable (field int)
        ''')
        cur.execute('select * from testtable')
        self.assertEqual([], cur.fetchall())

    def test_fixed_size_data(self):
        with self.conn.cursor() as cur:
            cur.execute('''
            create table testtable (chr char(5), nchr nchar(5), bfld binary(5))
            insert into testtable values ('1', '2', cast('3' as binary(5)))
            ''')
            cur.execute('select * from testtable')
            self.assertEqual(cur.fetchall(), [('1    ', '2    ', b'3\x00\x00\x00\x00')])


class TestVariant(TestCase):
    def _t(self, result, sql):
        with self.conn.cursor() as cur:
            cur.execute("select cast({} as sql_variant)".format(sql))
            val, = cur.fetchone()
            self.assertEqual(result, val)

    def test_new_datetime(self):
        if not IS_TDS73_PLUS(self.conn):
            self.skipTest('Requires TDS7.3+')
        self._t(datetime(2011, 2, 3, 10, 11, 12, 3000), "cast('2011-02-03T10:11:12.003000' as datetime2)")
        self._t(time(10, 11, 12, 3000), "cast('10:11:12.003000' as time)")
        self._t(date(2011, 2, 3), "cast('2011-02-03' as date)")
        self._t(datetime(2011, 2, 3, 10, 11, 12, 3000, tzoffset('', 3 * 60 * 60)), "cast('2011-02-03T10:11:12.003000+03:00' as datetimeoffset)")

    def test_regular(self):
        if not IS_TDS71_PLUS(self.conn):
            self.skipTest('Requires TDS7.1+')
        self._t(None, "cast(NULL as varchar)")
        self._t('test', "cast('test' as varchar)")
        self._t('test ', "cast('test' as char(5))")
        self._t('test', "cast(N'test' as nvarchar)")
        self._t('test ', "cast(N'test' as nchar(5))")
        self._t(Decimal('100.55555'), "cast(100.55555 as decimal(8,5))")
        self._t(Decimal('100.55555'), "cast(100.55555 as numeric(8,5))")
        self._t(b'test', "cast('test' as varbinary)")
        self._t(b'test\x00', "cast('test' as binary(5))")
        self._t(datetime(2011, 2, 3, 10, 11, 12, 3000), "cast('2011-02-03T10:11:12.003' as datetime)")
        self._t(datetime(2011, 2, 3, 10, 11, 0), "cast('2011-02-03T10:11:00' as smalldatetime)")
        val = uuid.uuid4()
        self._t(val, "cast('{}' as uniqueidentifier)".format(val))
        self._t(True, "cast(1 as bit)")
        self._t(128, "cast(128 as tinyint)")
        self._t(255, "cast(255 as tinyint)")
        self._t(-32000, "cast(-32000 as smallint)")
        self._t(2000000000, "cast(2000000000 as int)")
        self._t(2000000000000, "cast(2000000000000 as bigint)")
        self._t(0.12345, "cast(0.12345 as float)")
        self._t(0.25, "cast(0.25 as real)")
        self._t(Decimal('922337203685477.5807'), "cast('922,337,203,685,477.5807' as money)")
        self._t(Decimal('-214748.3648'), "cast('- 214,748.3648' as smallmoney)")


class BadConnection(unittest.TestCase):
    def test_invalid_parameters(self):
        with self.assertRaises(Error):
            with connect(server=settings.HOST, database='master', user=settings.USER, password=settings.PASSWORD + 'bad') as conn:
                with conn.cursor() as cur:
                    cur.execute('select 1')
        with self.assertRaises(Error):
            with connect(server=settings.HOST + 'bad', database='master', user=settings.USER + 'bad', password=settings.PASSWORD) as conn:
                with conn.cursor() as cur:
                    cur.execute('select 1')
        with self.assertRaises(Error):
            with connect(server=settings.HOST, database='doesnotexist', user=settings.USER, password=settings.PASSWORD) as conn:
                with conn.cursor() as cur:
                    cur.execute('select 1')
        with self.assertRaises(Error):
            with connect(server=settings.HOST, database='master', user=settings.USER, password=None) as conn:
                with conn.cursor() as cur:
                    cur.execute('select 1')

    def test_bad_instance_name(self):
        if not hasattr(settings, 'INSTANCE_PORT'):
            return self.skipTest('INSTANCE_PORT must be set to run this test')
        with self.assertRaisesRegexp(LoginError, 'Invalid instance name'):
            host = settings.HOST
            if '\\' in host:
                host, _ = host.split('\\')
            with connect(server=host + '\\badinstancename', database='master', user=settings.USER, password=settings.PASSWORD, port=settings.INSTANCE_PORT) as conn:
                with conn.cursor() as cur:
                    cur.execute('select 1')


def get_spid(conn):
    with conn.cursor() as cur:
        return cur.spid


def kill(conn, spid):
    with conn.cursor() as cur:
        cur.execute('kill {}'.format(spid))


class ConnectionClosing(unittest.TestCase):
    def test_open_close(self):
        for x in xrange(3):
            connect(server=settings.HOST, database='master', user=settings.USER, password=settings.PASSWORD).close()

    def test_closing_after_closed_by_server(self):
        '''
        You should be able to call close on connection closed by server
        '''
        with connect(server=settings.HOST,
                     database='master',
                     user=settings.USER,
                     password=<PASSWORD>,
                     autocommit=True) as master_conn:
            with connect(server=settings.HOST,
                         database='master',
                         user=settings.USER,
                         password=<PASSWORD>,
                         autocommit=False,
                         use_mars=settings.USE_MARS) as conn:
                with conn.cursor() as cur:
                    cur.execute('select 1')
                    conn.commit()
                    kill(master_conn, get_spid(conn))
                    sleep(0.2)
                conn.close()

    def test_connection_closed_by_server(self):
        with connect(server=settings.HOST,
                     database='master',
                     user=settings.USER,
                     password=<PASSWORD>,
                     autocommit=True) as master_conn:
            with connect(server=settings.HOST,
                         database='master',
                         user=settings.USER,
                         password=<PASSWORD>,
                         autocommit=False,
                         use_mars=settings.USE_MARS) as conn:
                # test overall recovery
                with conn.cursor() as cur:
                    cur.execute('select 1')
                    conn.commit()
                    kill(master_conn, get_spid(conn))
                    sleep(0.2)
                    cur.execute('select 1')
                    cur.fetchall()
                kill(master_conn, get_spid(conn))
                sleep(0.2)
                with conn.cursor() as cur:
                    cur.execute('select 1')

                # test recovery on transaction
                with conn.cursor() as cur:
                    cur.execute('create table ##testtable3 (fld int)')
                    kill(master_conn, get_spid(conn))
                    sleep(0.2)
                    with self.assertRaises(Exception):
                        cur.execute('select * from ##testtable2')
                        cur.fetchall()
                    conn.rollback()
                    cur.execute('select 1')
            #with connect(server=settings.HOST, database='master', user=settings.USER, password=settings.PASSWORD) as conn:
            #    spid = get_spid(conn)
            #    with conn.cursor() as cur:
            #        # test recovery of specific lowlevel methods
            #        tds_submit_query(cur._session, "waitfor delay '00:00:05'; select 1")
            #        kill(master_conn, spid)
            #        self.assertTrue(cur._session.is_connected())
            #        with self.assertRaises(Exception):
            #            tds_process_tokens(cur._session, TDS_TOKEN_RESULTS)
            #        self.assertFalse(cur._session.is_connected())


class Bug1(TestCase):
    def runTest(self):
        try:
            with connect(server=settings.HOST, database='master', user=settings.USER, password=settings.PASSWORD + '<PASSWORD>') as conn:
                with conn.cursor() as cur:
                    cur.execute('select 1')
                    cur.fetchall()
                conn.rollback()
        except:
            pass


#class EncryptionTest(unittest.TestCase):
#    def runTest(self):
#        conn = connect(server=settings.HOST, database='master', user=settings.USER, password=settings.PASSWORD, encryption_level=TDS_ENCRYPTION_REQUIRE)
#        cur = conn.cursor()
#        cur.execute('select 1')


class Bug2(DbTestCase):
    def runTest(self):
        with self.conn.cursor() as cur:
            cur.execute('''
            create procedure testproc (@param int)
            as
            begin
                set transaction isolation level read uncommitted -- that will produce very empty result (even no rowcount)
                select @param
                return @param + 1
            end
            ''')
            val = 45
            cur.execute('exec testproc @param = 45')
            self.assertEqual(cur.fetchall(), [(val,)])
            self.assertEqual(val + 1, cur.get_proc_return_status())


class Bug3(TestCase):
    def runTest(self):
        with self.conn.cursor() as cur:
            cur.close()


class DateAndTimeParams(TestCase):
    def test_date(self):
        if not IS_TDS73_PLUS(self.conn):
            self.skipTest('Requires TDS7.3+')
        with self.conn.cursor() as cur:
            date = Date(2012, 10, 6)
            cur.execute('select %s', (date, ))
            self.assertEqual(cur.fetchall(), [(date,)])

    def test_time(self):
        if not IS_TDS73_PLUS(self.conn):
            self.skipTest('Requires TDS7.3+')
        with self.conn.cursor() as cur:
            time = Time(8, 7, 4, 123000)
            cur.execute('select %s', (time, ))
            self.assertEqual(cur.fetchall(), [(time,)])

    def test_datetime(self):
        with self.conn.cursor() as cur:
            time = Timestamp(2013, 7, 9, 8, 7, 4, 123000)
            cur.execute('select %s', (time, ))
            self.assertEqual(cur.fetchall(), [(time,)])


class Extensions(TestCase):
    def runTest(self):
        with self.conn.cursor() as cur:
            self.assertEqual(cur.connection, self.conn)


class SmallDateTimeTest(TestCase):
    def _testval(self, val):
        with self.conn.cursor() as cur:
            cur.execute('select cast(%s as smalldatetime)', (val,))
            self.assertEqual(cur.fetchall(), [(val,)])

    def runTest(self):
        self._testval(Timestamp(2010, 2, 1, 10, 12, 0))
        self._testval(Timestamp(1900, 1, 1, 0, 0, 0))
        self._testval(Timestamp(2079, 6, 6, 23, 59, 0))
        with self.assertRaises(Error):
            self._testval(Timestamp(1899, 1, 1, 0, 0, 0))
        with self.assertRaises(Error):
            self._testval(Timestamp(2080, 1, 1, 0, 0, 0))


class DateTimeTest(DbTestCase):
    def _testencdec(self, val):
        self.assertEqual(val, DateTime.decode(*DateTime._struct.unpack(DateTime.encode(val))))

    def _testval(self, val):
        with self.conn.cursor() as cur:
            cur.execute('select cast(%s as datetime)', (val,))
            self.assertEqual(cur.fetchall(), [(val,)])

    def runTest(self):
        self.assertEqual(DateTime.decode(*DateTime._struct.unpack(b'\xf2\x9c\x00\x00}uO\x01')), Timestamp(2010, 1, 2, 20, 21, 22, 123000))
        self.assertEqual(DateTime.decode(*DateTime._struct.unpack(b'\x7f$-\x00\xff\x81\x8b\x01')), DateTime._max_date)
        self.assertEqual(b'\xf2\x9c\x00\x00}uO\x01', DateTime.encode(Timestamp(2010, 1, 2, 20, 21, 22, 123000)))
        self.assertEqual(b'\x7f$-\x00\xff\x81\x8b\x01', DateTime.encode(DateTime._max_date))
        with self.conn.cursor() as cur:
            cur.execute("select cast('9999-12-31T23:59:59.997' as datetime)")
            self.assertEqual(cur.fetchall(), [(Timestamp(9999, 12, 31, 23, 59, 59, 997000),)])
        self._testencdec(Timestamp(2010, 1, 2, 10, 11, 12))
        self._testval(Timestamp(2010, 1, 2, 0, 0, 0))
        self._testval(Timestamp(2010, 1, 2, 10, 11, 12))
        self._testval(Timestamp(1753, 1, 1, 0, 0, 0))
        self._testval(Timestamp(9999, 12, 31, 0, 0, 0))
        with self.conn.cursor() as cur:
            cur.execute("select cast(null as datetime)")
            self.assertEqual(cur.fetchall(), [(None,)])
        self._testval(Timestamp(9999, 12, 31, 23, 59, 59, 997000))
        with self.assertRaises(Error):
            self._testval(Timestamp(1752, 1, 1, 0, 0, 0))
        with self.conn.cursor() as cur:
            cur.execute('''
            if object_id('testtable') is not null
                drop table testtable
            ''')
            cur.execute('create table testtable (col datetime not null)')
            dt = Timestamp(2010, 1, 2, 20, 21, 22, 123000)
            cur.execute('insert into testtable values (%s)', (dt,))
            cur.execute('select col from testtable')
            self.assertEqual(cur.fetchone(), (dt,))


class NewDateTimeTest(TestCase):
    def test_datetimeoffset(self):
        if not IS_TDS73_PLUS(self.conn):
            self.skipTest('Requires TDS7.3+')

        def _testval(val):
            with self.conn.cursor() as cur:
                cur.execute('select cast(%s as datetimeoffset)', (val,))
                self.assertEqual(cur.fetchall(), [(val,)])

        with self.conn.cursor() as cur:
            cur.execute("select cast('2010-01-02T20:21:22.1234567+05:00' as datetimeoffset)")
            self.assertEqual(datetime(2010, 1, 2, 20, 21, 22, 123456, tzoffset('', 5 * 60 * 60)), cur.fetchone()[0])
        _testval(Timestamp(2010, 1, 2, 0, 0, 0, 0, tzutc()))
        _testval(Timestamp(2010, 1, 2, 0, 0, 0, 0, tzoffset('', 5 * 60 * 60)))
        _testval(Timestamp(1, 1, 1, 0, 0, 0, 0, tzutc()))
        _testval(Timestamp(9999, 12, 31, 23, 59, 59, 999999, tzutc()))
        _testval(Timestamp(2010, 1, 2, 0, 0, 0, 0, tzoffset('', 14 * 60)))
        _testval(Timestamp(2010, 1, 2, 0, 0, 0, 0, tzoffset('', -14 * 60)))
        _testval(Timestamp(2010, 1, 2, 0, 0, 0, 0, tzoffset('', -15 * 60)))

    def test_time(self):
        if not IS_TDS73_PLUS(self.conn):
            self.skipTest('Requires TDS7.3+')

        def testval(val):
            with self.conn.cursor() as cur:
                cur.execute('select cast(%s as time)', (val,))
                self.assertEqual(cur.fetchall(), [(val,)])

        testval(Time(14, 16, 18, 123456))
        testval(Time(0, 0, 0, 0))
        testval(Time(0, 0, 0, 0))
        testval(Time(0, 0, 0, 0))
        testval(Time(23, 59, 59, 999999))
        testval(Time(0, 0, 0, 0))
        testval(Time(0, 0, 0, 0))
        testval(Time(0, 0, 0, 0))

    def test_datetime2(self):
        if not IS_TDS73_PLUS(self.conn):
            self.skipTest('Requires TDS7.3+')

        def testval(val):
            with self.conn.cursor() as cur:
                cur.execute('select cast(%s as datetime2)', (val,))
                self.assertEqual(cur.fetchall(), [(val,)])

        testval(Timestamp(2010, 1, 2, 20, 21, 22, 345678))
        testval(Timestamp(2010, 1, 2, 0, 0, 0))
        testval(Timestamp(1, 1, 1, 0, 0, 0))
        testval(Timestamp(9999, 12, 31, 23, 59, 59, 999999))

    def test_date(self):
        if not IS_TDS73_PLUS(self.conn):
            self.skipTest('Requires TDS7.3+')

        def testval(val):
            with self.conn.cursor() as cur:
                cur.execute('select cast(%s as date)', (val,))
                self.assertEqual(cur.fetchall(), [(val,)])

        testval(Date(2010, 1, 2))
        testval(Date(2010, 1, 2))
        testval(Date(1, 1, 1))
        testval(Date(9999, 12, 31))


class Auth(unittest.TestCase):
    #def test_ntlm(self):
    #    conn = connect(settings.HOST, auth=NtlmAuth(user_name=settings.NTLM_USER, password=settings.NTLM_PASSWORD))
    #    with conn.cursor() as cursor:
    #        cursor.execute('select 1')
    #        cursor.fetchall()

    @unittest.skipUnless(sys.platform.startswith("win"), "requires Windows")
    def test_sspi(self):
        from pytds.login import SspiAuth
        with connect(settings.HOST, auth=SspiAuth()) as conn:
            with conn.cursor() as cursor:
                cursor.execute('select 1')
                cursor.fetchall()

    def test_sqlauth(self):
        with connect(settings.HOST, user=settings.USER, password=settings.PASSWORD) as conn:
            with conn.cursor() as cursor:
                cursor.execute('select 1')
                cursor.fetchall()


class CloseCursorTwice(TestCase):
    def runTest(self):
        cursor = self.conn.cursor()
        cursor.close()
        cursor.close()


class RegressionSuite(TestCase):
    def test_commit(self):
        if self.conn.mars_enabled:
            self.skipTest('Only breaks when mars is disabled')
        cursor = self.conn.cursor()
        cursor.execute('select 1')
        cursor.fetchall()
        self.conn.commit()
<commit_msg>added SKIP_SQL_AUTH flag<commit_after># vim: set fileencoding=utf8 :
from __future__ import with_statement
from six import StringIO
import unittest
import sys
from decimal import Decimal, getcontext
import logging
from time import sleep
from datetime import datetime, date, time
import uuid
from dateutil.tz import tzoffset, tzutc
from six import text_type
from six.moves import xrange
import binascii
from pytds import (
    connect, ProgrammingError, TimeoutError, Time, SimpleLoadBalancer, LoginError,
    Error, IntegrityError, Timestamp, DataError, DECIMAL, Date, Binary, DateTime,
    IS_TDS73_PLUS, IS_TDS71_PLUS, NotSupportedError, TDS73, TDS71, TDS72,
    output, default, InterfaceError, TDS_ENCRYPTION_OFF)
from pytds.tds import (
    _TdsSocket, _TdsSession, TDS_ENCRYPTION_REQUIRE, Column, Bit, Int, SmallInt,
    NVarChar72, TinyInt, IntN, BigInt, Real, Float, FloatN, Collation,
    )
from pytds.dbapi import _TdsLogin
from . import dbapi20
import pytds

# set decimal precision to match mssql maximum precision
getcontext().prec = 38

try:
    from . import settings
except:
    print('Settings module is not found, please create settings module and specify HOST, DATATABSE, USER and PASSWORD there')
    sys.exit(1)

#logging.basicConfig(level='DEBUG')
#logging.basicConfig(level='INFO')
logging.basicConfig()


class TestCase(unittest.TestCase):
    def setUp(self):
        kwargs = settings.CONNECT_KWARGS.copy()
        kwargs['database'] = 'master'
        self.conn = connect(*settings.CONNECT_ARGS, **kwargs)

    def tearDown(self):
        self.conn.close()


class DbTestCase(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        kwargs = settings.CONNECT_KWARGS.copy()
        kwargs['database'] = 'master'
        kwargs['autocommit'] = True
        with connect(**kwargs) as conn:
            with conn.cursor() as cur:
                try:
                    cur.execute('drop database [{}]'.format(settings.DATABASE))
                except:
                    pass
                cur.execute('create database [{}]'.format(settings.DATABASE))

    @classmethod
    def tearDownClass(cls):
        kwargs = settings.CONNECT_KWARGS.copy()
        kwargs['database'] = 'master'
        kwargs['autocommit'] = True
        with connect(**kwargs) as conn:
            with conn.cursor() as cur:
                cur.execute('drop database [{}]'.format(settings.DATABASE))

    def setUp(self):
        self.conn = pytds.connect(*settings.CONNECT_ARGS, **settings.CONNECT_KWARGS)

    def tearDown(self):
        self.conn.close()


class TestCase2(TestCase):
    def test_all(self):
        cur = self.conn.cursor()
        with self.assertRaises(ProgrammingError):
            cur.execute(u'select ')
        assert 'abc' == cur.execute_scalar("select cast('abc' as nvarchar(max)) as fieldname")
        assert b'abc' == cur.execute_scalar("select cast('abc' as varbinary(max)) as fieldname")
        #assert 12 == cur.execute_scalar('select cast(12 as bigint) as fieldname')
        assert 12 == cur.execute_scalar('select cast(12 as smallint) as fieldname')
        assert -12 == cur.execute_scalar('select -12 as fieldname')
        assert 12 == cur.execute_scalar('select cast(12 as tinyint) as fieldname')
        assert True == cur.execute_scalar('select cast(1 as bit) as fieldname')
        assert 5.1 == cur.execute_scalar('select cast(5.1 as float) as fieldname')
        cur.execute("select 'test', 20")
        assert ('test', 20) == cur.fetchone()
        assert 'test' == cur.execute_scalar("select 'test' as fieldname")
        assert 'test' == cur.execute_scalar("select N'test' as fieldname")
        assert 'test' == cur.execute_scalar("select cast(N'test' as ntext) as fieldname")
        assert 'test' == cur.execute_scalar("select cast(N'test' as text) as fieldname")
        self.assertEqual('test ', cur.execute_scalar("select cast(N'test' as char(5)) as fieldname"))
        self.assertEqual('test ', cur.execute_scalar("select cast(N'test' as nchar(5)) as fieldname"))
        assert b'test' == cur.execute_scalar("select cast('test' as varbinary(4)) as fieldname")
        assert b'test' == cur.execute_scalar("select cast('test' as image) as fieldname")
        assert None == cur.execute_scalar("select cast(NULL as image) as fieldname")
        assert None == cur.execute_scalar("select cast(NULL as varbinary(10)) as fieldname")
        assert None == cur.execute_scalar("select cast(NULL as ntext) as fieldname")
        assert None == cur.execute_scalar("select cast(NULL as nvarchar(max)) as fieldname")
        assert None == cur.execute_scalar("select cast(NULL as xml)")
        self.assertEqual(None, cur.execute_scalar("select cast(NULL as varchar(max)) as fieldname"))
        assert None == cur.execute_scalar("select cast(NULL as nvarchar(10)) as fieldname")
        assert None == cur.execute_scalar("select cast(NULL as varchar(10)) as fieldname")
        assert None == cur.execute_scalar("select cast(NULL as nchar(10)) as fieldname")
        assert None == cur.execute_scalar("select cast(NULL as char(10)) as fieldname")
        assert None == cur.execute_scalar("select cast(NULL as char(10)) as fieldname")
        self.assertEqual(u'Iñtërnâtiônàlizætiøn1', cur.execute_scalar('select %s', (u'Iñtërnâtiônàlizætiøn1'.encode('utf8'),)))
        assert 5 == cur.execute_scalar('select 5 as fieldname')

    def test_decimals(self):
        cur = self.conn.cursor()
        assert Decimal(12) == cur.execute_scalar('select cast(12 as decimal) as fieldname')
        assert Decimal(-12) == cur.execute_scalar('select cast(-12 as decimal) as fieldname')
        assert Decimal('123456.12345') == cur.execute_scalar("select cast('123456.12345'as decimal(20,5)) as fieldname")
        assert Decimal('-123456.12345') == cur.execute_scalar("select cast('-123456.12345'as decimal(20,5)) as fieldname")

    def test_money(self):
        cur = self.conn.cursor()
        assert Decimal('0') == cur.execute_scalar("select cast('0' as money) as fieldname")
        assert Decimal('1') == cur.execute_scalar("select cast('1' as money) as fieldname")
        self.assertEqual(Decimal('1.5555'), cur.execute_scalar("select cast('1.5555' as money) as fieldname"))
        assert Decimal('1234567.5555') == cur.execute_scalar("select cast('1234567.5555' as money) as fieldname")
        assert Decimal('-1234567.5555') == cur.execute_scalar("select cast('-1234567.5555' as money) as fieldname")
        assert Decimal('12345.55') == cur.execute_scalar("select cast('12345.55' as smallmoney) as fieldname")

    def test_timeout(self):
        kwargs = settings.CONNECT_KWARGS.copy()
        kwargs['database'] = 'master'
        kwargs['login_timeout'] = 1
        with connect(*settings.CONNECT_ARGS, **kwargs) as conn:
            cur = conn.cursor()
            with self.assertRaises(TimeoutError):
                cur.execute("waitfor delay '00:00:05'")
            cur.execute('select 1')

    def test_strs(self):
        cur = self.conn.cursor()
        self.assertIsInstance(cur.execute_scalar("select 'test'"), text_type)

    #def test_mars_sessions_recycle_ids(self):
    #    if not self.conn.mars_enabled:
    #        self.skipTest('Only relevant to mars')
    #    for _ in xrange(2 ** 16 + 1):
    #        cur = self.conn.cursor()
    #        cur.close()

    def test_smp(self):
        if not self.conn.mars_enabled:
            self.skipTest('Only relevant to mars')
        sess = self.conn._conn._smp_manager.create_session()
        self.assertEqual(sess._state, 'SESSION ESTABLISHED')
        sess.close()
        self.assertEqual(sess._state, 'CLOSED')

    def test_cursor_env(self):
        with self.conn.cursor() as cur:
            cur.execute('use master')
            self.assertEqual(cur.execute_scalar('select DB_NAME()'), 'master')

    def test_empty_query(self):
        with self.conn.cursor() as cur:
            cur.execute('')
            self.assertIs(None, cur.description)

    def test_parameters_ll(self):
        def test_val(typ, val):
            with self.conn.cursor() as cur:
                param = Column(type=typ, value=val)
                cur.execute('select %s', [param])
                self.assertTupleEqual(cur.fetchone(), (val,))
                self.assertIs(cur.fetchone(), None)

        test_val(self.conn._conn.BitN, True)
        test_val(self.conn._conn.BitN, False)
        test_val(self.conn._conn.BitN, None)
        test_val(self.conn._conn.IntN(1), 255)
        test_val(self.conn._conn.IntN(2), 2 ** 15 - 1)
        test_val(self.conn._conn.IntN(4), 2 ** 31 - 1)
        test_val(self.conn._conn.IntN(8), 2 ** 63 - 1)
        test_val(self.conn._conn.IntN(4), None)
        #test_val(self.conn._conn.Real, 0.25)
        #test_val(self.conn._conn.Float, 0.25)
        test_val(self.conn._conn.FloatN(4), 0.25)
        test_val(self.conn._conn.FloatN(8), 0.25)
        test_val(self.conn._conn.FloatN(4), None)
        #test_val(self.conn._conn.SmallDateTime, datetime(1900, 1, 1, 0, 0, 0))
        #test_val(self.conn._conn.SmallDateTime, datetime(2079, 6, 6, 23, 59, 0))
        #test_val(self.conn._conn.DateTime, datetime(1753, 1, 1, 0, 0, 0))
        #test_val(self.conn._conn.DateTime, datetime(9999, 12, 31, 23, 59, 59, 990000))
        test_val(self.conn._conn.DateTimeN(4), datetime(1900, 1, 1, 0, 0, 0))
        test_val(self.conn._conn.DateTimeN(4), datetime(2079, 6, 6, 23, 59, 0))
        test_val(self.conn._conn.DateTimeN(8), datetime(1753, 1, 1, 0, 0, 0))
        test_val(self.conn._conn.DateTimeN(8), datetime(9999, 12, 31, 23, 59, 59, 990000))
        test_val(self.conn._conn.DateTimeN(8), None)
        test_val(self.conn._conn.Date, date(1, 1, 1))
        test_val(self.conn._conn.Date, date(9999, 12, 31))
        test_val(self.conn._conn.Date, None)
        test_val(self.conn._conn.Time(0), time(0, 0, 0))
        test_val(self.conn._conn.Time(6), time(23, 59, 59, 999999))
        test_val(self.conn._conn.Time(0), None)
        test_val(self.conn._conn.DateTime2(0), datetime(1, 1, 1, 0, 0, 0))
        test_val(self.conn._conn.DateTime2(6), datetime(9999, 12, 31, 23, 59, 59, 999999))
        test_val(self.conn._conn.DateTime2(0), None)
        test_val(self.conn._conn.DateTimeOffset(6), datetime(9999, 12, 31, 23, 59, 59, 999999, tzutc()))
        test_val(self.conn._conn.DateTimeOffset(6), datetime(9999, 12, 31, 23, 59, 59, 999999, tzoffset('', 14 * 60)))
        test_val(self.conn._conn.DateTimeOffset(0), datetime(1, 1, 1, 0, 0, 0, tzinfo=tzoffset('', -14 * 60)))
        #test_val(self.conn._conn.DateTimeOffset(0), datetime(1, 1, 1, 0, 0, 0, tzinfo=tzoffset('', 14 * 60)))
        test_val(self.conn._conn.DateTimeOffset(6), None)
        test_val(self.conn._conn.Decimal(6, 38), Decimal('123.456789'))
        test_val(self.conn._conn.Decimal(6, 38), None)
        #test_val(self.conn._conn.SmallMoney, Decimal('214748.3647'))
        #test_val(self.conn._conn.SmallMoney, Decimal('-214748.3648'))
        #test_val(self.conn._conn.Money, Decimal('922337203685477.5807'))
        #test_val(self.conn._conn.Money, Decimal('-922337203685477.5808'))
        test_val(self.conn._conn.MoneyN(4), Decimal('-214748.3648'))
        test_val(self.conn._conn.MoneyN(4), Decimal('214748.3647'))
        test_val(self.conn._conn.MoneyN(8), Decimal('922337203685477.5807'))
        test_val(self.conn._conn.MoneyN(8), Decimal('-922337203685477.5808'))
        test_val(self.conn._conn.MoneyN(8), None)
        test_val(self.conn._conn.UniqueIdentifier, None)
        test_val(self.conn._conn.UniqueIdentifier, uuid.uuid4())
        test_val(self.conn._conn.SqlVariant(10), None)
        #test_val(self.conn._conn.SqlVariant(10), 100)
        test_val(self.conn._conn.VarBinary(10), b'')
        test_val(self.conn._conn.VarBinary(10), b'testtest12')
        test_val(self.conn._conn.VarBinary(10), None)
        test_val(self.conn._conn.VarBinary(8000), b'x' * 8000)
        test_val(self.conn._conn.long_binary_type(), None)
        test_val(self.conn._conn.long_binary_type(), b'')
        test_val(self.conn._conn.long_binary_type(), b'testtest12')
        test_val(self.conn._conn.long_binary_type(), b'x' * (10 ** 6))
        test_val(self.conn._conn.NVarChar(10), u'')
        test_val(self.conn._conn.NVarChar(10), u'testtest12')
        test_val(self.conn._conn.NVarChar(10), None)
        test_val(self.conn._conn.NVarChar(4000), u'x' * 4000)
        test_val(self.conn._conn.long_string_type(), None)
        test_val(self.conn._conn.long_string_type(), 'test')
        test_val(self.conn._conn.long_string_type(), 'x' * (10 ** 6))
        #test_val(self.conn._conn.Xml(), '<root></root>')
        #test_val(self.conn._conn.Image(10, []), None)
        #test_val(self.conn._conn.Image(10, ['']), None)
        #test_val(self.conn._conn.Image(10, ['']), b'test')

    def test_parameters(self):
        def test_val(val):
            with self.conn.cursor() as cur:
                cur.execute('select %s', (val,))
                self.assertTupleEqual(cur.fetchone(), (val,))
                self.assertIs(cur.fetchone(), None)

        test_val(u'hello')
        test_val(123)
        test_val(-123)
        test_val(123.12)
        test_val(-123.12)
        test_val(10 ** 20)
        test_val(10 ** 38 - 1)
        test_val(-10 ** 38 + 1)
        test_val(datetime(2011, 2, 3, 10, 11, 12, 3000))
        test_val(Decimal('1234.567'))
        test_val(Decimal('1234000'))
        test_val(Decimal('9' * 38))
        test_val(Decimal('0.' + '9' * 38))
        test_val(-Decimal('9' * 38))
        test_val(Decimal('1E10'))
        test_val(Decimal('1E-10'))
        test_val(Decimal('0.{0}1'.format('0' * 37)))
        test_val(None)
        test_val('hello')
        test_val('')
        test_val('x' * 5000)
        test_val(Binary(b'\x00\x01\x02'))
        test_val(Binary(b'x' * 5000))
        test_val(2 ** 63 - 1)
        test_val(False)
        test_val(True)
        test_val(uuid.uuid4())

    def test_overlimit(self):
        def test_val(val):
            with self.conn.cursor() as cur:
                cur.execute('select %s', (val,))
                self.assertTupleEqual(cur.fetchone(), (val,))
                self.assertIs(cur.fetchone(), None)

        with self.assertRaises(DataError):
            test_val(-10 ** 38)
        ##cur.execute('select %s', '\x00'*(2**31))
        with self.assertRaises(DataError):
            test_val(Decimal('1' + '0' * 38))
        with self.assertRaises(DataError):
            test_val(Decimal('-1' + '0' * 38))
        with self.assertRaises(DataError):
            test_val(Decimal('1E38'))

    def test_description(self):
        with self.conn.cursor() as cur:
            cur.execute('select cast(12.65 as decimal(4,2)) as testname')
            self.assertEqual(cur.description[0][0], 'testname')
            self.assertEqual(cur.description[0][1], DECIMAL)
            self.assertEqual(cur.description[0][4], 4)
            self.assertEqual(cur.description[0][5], 2)

    def test_bug4(self):
        with self.conn.cursor() as cur:
            cur.execute('''
            set transaction isolation level read committed
            select 1
            ''')
            self.assertEqual(cur.fetchall(), [(1,)])

    def test_bad_collation(self):
        with self.conn.cursor() as cur:
            try:
                cur.execute_scalar('select cast(0x90 as varchar)')
            except:
                pass
            self.assertEqual(1, cur.execute_scalar('select 1'))

    def test_get_instances(self):
        if not hasattr(settings, 'BROWSER_ADDRESS'):
            return unittest.skip('BROWSER_ADDRESS setting is not defined')
        pytds.tds.tds7_get_instances(settings.BROWSER_ADDRESS)


class DbTests(DbTestCase):
    def test_autocommit(self):
        self.assertFalse(self.conn.autocommit)
        with self.conn.cursor() as cur:
            cur.execute('create table test_autocommit(field int)')
            self.conn.commit()
            self.assertEqual(self.conn._trancount(), 1)
            cur.execute('insert into test_autocommit(field) values(1)')
            self.assertEqual(self.conn._trancount(), 1)
            cur.execute('select field from test_autocommit')
            row = cur.fetchone()
            self.conn.rollback()
            cur.execute('select field from test_autocommit')
            row = cur.fetchone()
            self.assertFalse(row)

            self.conn.autocommit = True
            cur.execute('insert into test_autocommit(field) values(1)')
            self.assertEqual(self.conn._trancount(), 0)

    def _test_bulk_type(self, typ, value):
        with self.conn.cursor() as cur:
            cur.execute('create table bulk_insert_table_ll(c1 {})'.format(typ.get_declaration()))
            cur._session.submit_plain_query('insert bulk bulk_insert_table_ll (c1 {})'.format(typ.get_declaration()))
            cur._session.process_simple_request()
            col1 = Column('c1', typ, flags=Column.fNullable)
            metadata = [col1]
            cur._session.submit_bulk(metadata, [(value,)])
            cur._session.process_simple_request()
        with self.conn.cursor() as cur:
            cur.execute('select c1 from bulk_insert_table_ll')
            self.assertTupleEqual(cur.fetchone(), (value,))
            self.assertIs(cur.fetchone(), None)
            cur.execute('drop table bulk_insert_table_ll')

    def test_bulk_insert_low_level(self):
        self._test_bulk_type(Bit.instance, True)
        self._test_bulk_type(Bit.instance, False)
        self._test_bulk_type(Int.instance, 2 ** 31 - 1)
        self._test_bulk_type(Int.instance, -2 ** 31)
        self._test_bulk_type(SmallInt.instance, -2 ** 15)
        self._test_bulk_type(SmallInt.instance, 2 ** 15 - 1)
        self._test_bulk_type(TinyInt.instance, 255)
        self._test_bulk_type(TinyInt.instance, 0)
        self._test_bulk_type(BigInt.instance, 2 ** 63 - 1)
        self._test_bulk_type(BigInt.instance, -2 ** 63)
        self._test_bulk_type(IntN(1), 255)
        self._test_bulk_type(IntN(2), 2 ** 15 - 1)
        self._test_bulk_type(IntN(4), 2 ** 31 - 1)
        self._test_bulk_type(IntN(8), 2 ** 63 - 1)
        self._test_bulk_type(IntN(4), None)
        self._test_bulk_type(Real.instance, 0.25)
        self._test_bulk_type(Float.instance, 0.25)
        self._test_bulk_type(FloatN(4), 0.25)
        self._test_bulk_type(FloatN(8), 0.25)
        self._test_bulk_type(FloatN(4), None)
        self._test_bulk_type(self.conn._conn.NVarChar(10), u'')
        self._test_bulk_type(self.conn._conn.NVarChar(10), u'testtest12')
        self._test_bulk_type(self.conn._conn.NVarChar(10), None)
        self._test_bulk_type(self.conn._conn.NVarChar(4000), u'x' * 4000)
        self._test_bulk_type(self.conn._conn.VarBinary(10), b'')
        self._test_bulk_type(self.conn._conn.VarBinary(10), b'testtest12')
        self._test_bulk_type(self.conn._conn.VarBinary(10), None)
        self._test_bulk_type(self.conn._conn.VarBinary(8000), b'x' * 8000)
        self._test_bulk_type(self.conn._conn.SmallDateTime, datetime(1900, 1, 1, 0, 0, 0))
        self._test_bulk_type(self.conn._conn.SmallDateTime, datetime(2079, 6, 6, 23, 59, 0))
        self._test_bulk_type(self.conn._conn.DateTime, datetime(1753, 1, 1, 0, 0, 0))
        self._test_bulk_type(self.conn._conn.DateTime, datetime(9999, 12, 31, 23, 59, 59, 990000))
        self._test_bulk_type(self.conn._conn.DateTimeN(4), datetime(1900, 1, 1, 0, 0, 0))
        self._test_bulk_type(self.conn._conn.DateTimeN(8), datetime(9999, 12, 31, 23, 59, 59, 990000))
        self._test_bulk_type(self.conn._conn.DateTimeN(8), None)
        self._test_bulk_type(self.conn._conn.Date, date(1, 1, 1))
        self._test_bulk_type(self.conn._conn.Date, date(9999, 12, 31))
        self._test_bulk_type(self.conn._conn.Date, None)
        self._test_bulk_type(self.conn._conn.Time(0), time(0, 0, 0))
        self._test_bulk_type(self.conn._conn.Time(6), time(23, 59, 59, 999999))
        self._test_bulk_type(self.conn._conn.Time(0), None)
        self._test_bulk_type(self.conn._conn.DateTime2(0), datetime(1, 1, 1, 0, 0, 0))
        self._test_bulk_type(self.conn._conn.DateTime2(6), datetime(9999, 12, 31, 23, 59, 59, 999999))
        self._test_bulk_type(self.conn._conn.DateTime2(0), None)
        self._test_bulk_type(self.conn._conn.DateTimeOffset(6), datetime(9999, 12, 31, 23, 59, 59, 999999, tzutc()))
        self._test_bulk_type(self.conn._conn.DateTimeOffset(6), datetime(9999, 12, 31, 23, 59, 59, 999999, tzoffset('', 14 * 60)))
        self._test_bulk_type(self.conn._conn.DateTimeOffset(0), datetime(1, 1, 1, 0, 0, 0, tzinfo=tzoffset('', -14 * 60)))
        #self._test_bulk_type(self.conn._conn.DateTimeOffset(0), datetime(1, 1, 1, 0, 0, 0, tzinfo=tzoffset('', 14 * 60)))
        self._test_bulk_type(self.conn._conn.DateTimeOffset(6), None)
        self._test_bulk_type(self.conn._conn.Decimal(6, 38), Decimal('123.456789'))
        self._test_bulk_type(self.conn._conn.Decimal(6, 38), None)
        self._test_bulk_type(self.conn._conn.SmallMoney, Decimal('214748.3647'))
        self._test_bulk_type(self.conn._conn.SmallMoney, Decimal('-214748.3648'))
        self._test_bulk_type(self.conn._conn.Money, Decimal('922337203685477.5807'))
        self._test_bulk_type(self.conn._conn.Money, Decimal('-922337203685477.5808'))
        self._test_bulk_type(self.conn._conn.MoneyN(4), Decimal('214748.3647'))
        self._test_bulk_type(self.conn._conn.MoneyN(8), Decimal('922337203685477.5807'))
        self._test_bulk_type(self.conn._conn.MoneyN(8), None)
        self._test_bulk_type(self.conn._conn.UniqueIdentifier, None)
        self._test_bulk_type(self.conn._conn.UniqueIdentifier, uuid.uuid4())
        self._test_bulk_type(self.conn._conn.SqlVariant(10), None)
        #self._test_bulk_type(self.conn._conn.SqlVariant(10), 100)
        self._test_bulk_type(self.conn._conn.long_binary_type(), None)
        #self._test_bulk_type(self.conn._conn.long_binary_type(), b'')
        #self._test_bulk_type(self.conn._conn.long_binary_type(), b'testtest12')
        self._test_bulk_type(self.conn._conn.long_string_type(), None)
        #self._test_bulk_type(self.conn._conn.long_string_type(), 'test')
        #self._test_bulk_type(self.conn._conn.Image(10, []), None)
        #self._test_bulk_type(self.conn._conn.Image(10, ['']), None)
        #self._test_bulk_type(self.conn._conn.Image(10, ['']), b'test')

    #def test_bulk_insert(self):
    #    with self.conn.cursor() as cur:
    #        cur.execute('create table bulk_insert_table(num int, data varchar(100))')
    #        f = StringIO("42\tfoo\n74\tbar\n")
    #        cur.copy_to(f, 'bulk_insert_table', columns=('num', 'data'))
    #        cur.execute('select num, data from bulk_insert_table')
    #        self.assertListEqual(cur.fetchall(), [(42, 'foo'), (74, 'bar')])

    def test_table_selects(self):
        cur = self.conn.cursor()
        cur.execute(u'''
        create table #testtable (id int, _text text, _xml xml, vcm varchar(max), vc varchar(10))
        ''')
        cur.execute(u'''
        insert into #testtable (id, _text, _xml, vcm, vc) values (1, 'text', '<root/>', '', NULL)
        ''')
        cur.execute('select id from #testtable order by id')
        self.assertEqual([(1,)], cur.fetchall())

        cur = self.conn.cursor()
        cur.execute('select _text from #testtable order by id')
        self.assertEqual([(u'text',)], cur.fetchall())

        cur = self.conn.cursor()
        cur.execute('select _xml from #testtable order by id')
        self.assertEqual([('<root/>',)], cur.fetchall())

        cur = self.conn.cursor()
        cur.execute('select id, _text, _xml, vcm, vc from #testtable order by id')
        self.assertTupleEqual((1, 'text', '<root/>', '', None), cur.fetchone())

        cur = self.conn.cursor()
        cur.execute('select vc from #testtable order by id')
        self.assertEqual([(None,)], cur.fetchall())

        cur = self.conn.cursor()
        cur.execute('insert into #testtable (_xml) values (%s)', ('<some/>',))

        cur = self.conn.cursor()
        cur.execute(u'drop table #testtable')

    def test_stored_proc(self):
        cur = self.conn.cursor()
        cur.execute('''
        create procedure testproc (@param int, @add int = 2, @outparam int output)
        as
        begin
            set nocount on
            --select @param
            set @outparam = @param + @add
            return @outparam
        end
        ''')
        val = 45
        #params = {'@param': val, '@outparam': output(None), '@add': 1}
        values = cur.callproc('testproc', (val, default, output(None, 1)))
        #self.assertEqual(cur.fetchall(), [(val,)])
        self.assertEqual(val + 2, values[2])
        self.assertEqual(val + 2, cur.get_proc_return_status())

    def test_fetchone(self):
        with self.conn.cursor() as cur:
            cur.execute('select 10; select 12')
            self.assertEqual((10,), cur.fetchone())
            self.assertTrue(cur.nextset())
            self.assertEqual((12,), cur.fetchone())
            self.assertFalse(cur.nextset())

    def test_fetchall(self):
        with self.conn.cursor() as cur:
            cur.execute('select 10; select 12')
            self.assertEqual([(10,)], cur.fetchall())
            self.assertTrue(cur.nextset())
            self.assertEqual([(12,)], cur.fetchall())
            self.assertFalse(cur.nextset())

    def test_cursor_closing(self):
        with self.conn.cursor() as cur:
            cur.execute('select 10; select 12')
            cur.fetchone()
        with self.conn.cursor() as cur2:
            cur2.execute('select 20')
            cur2.fetchone()

    def test_transactions(self):
        self.conn.autocommit = False
        with self.conn.cursor() as cur:
            cur.execute('''
            create table testtable (field datetime)
            ''')
            cur.execute("select object_id('testtable')")
            self.assertNotEqual((None,), cur.fetchone())

            self.assertEqual(1, self.conn._trancount())

            self.conn.rollback()

            self.assertEqual(1, self.conn._trancount())

            cur.execute("select object_id('testtable')")
            self.assertEqual((None,), cur.fetchone())

            cur.execute('''
            create table testtable (field datetime)
            ''')

            self.conn.commit()

            cur.execute("select object_id('testtable')")
            self.assertNotEqual((None,), cur.fetchone())

        with self.conn.cursor() as cur:
            cur.execute('''
            if object_id('testtable') is not null
                drop table testtable
            ''')
        self.conn.commit()

    def test_multi_packet(self):
        cur = self.conn.cursor()
        param = 'x' * (self.conn._conn.main_session._writer.bufsize * 3)
        cur.execute('select %s', (param,))
        self.assertEqual([(param, )], cur.fetchall())

    def test_big_request(self):
        with self.conn.cursor() as cur:
            param = 'x' * 5000
            params = (10, datetime(2012, 11, 19, 1, 21, 37, 3000), param, 'test')
            cur.execute('select %s, %s, %s, %s', params)
            self.assertEqual([params], cur.fetchall())

    def test_row_count(self):
        cur = self.conn.cursor()
        cur.execute('''
        create table testtable (field int)
        ''')
        cur.execute('insert into testtable (field) values (1)')
        self.assertEqual(cur.rowcount, 1)
        cur.execute('insert into testtable (field) values (2)')
        self.assertEqual(cur.rowcount, 1)
        cur.execute('select * from testtable')
        cur.fetchall()
        self.assertEqual(cur.rowcount, 2)

    def test_no_rows(self):
        cur = self.conn.cursor()
        cur.execute('''
        create table testtable (field int)
        ''')
        cur.execute('select * from testtable')
        self.assertEqual([], cur.fetchall())

    def test_fixed_size_data(self):
        with self.conn.cursor() as cur:
            cur.execute('''
            create table testtable (chr char(5), nchr nchar(5), bfld binary(5))
            insert into testtable values ('1', '2', cast('3' as binary(5)))
            ''')
            cur.execute('select * from testtable')
            self.assertEqual(cur.fetchall(), [('1    ', '2    ', b'3\x00\x00\x00\x00')])


class TestVariant(TestCase):
    def _t(self, result, sql):
        with self.conn.cursor() as cur:
            cur.execute("select cast({} as sql_variant)".format(sql))
            val, = cur.fetchone()
            self.assertEqual(result, val)

    def test_new_datetime(self):
        if not IS_TDS73_PLUS(self.conn):
            self.skipTest('Requires TDS7.3+')
        self._t(datetime(2011, 2, 3, 10, 11, 12, 3000), "cast('2011-02-03T10:11:12.003000' as datetime2)")
        self._t(time(10, 11, 12, 3000), "cast('10:11:12.003000' as time)")
        self._t(date(2011, 2, 3), "cast('2011-02-03' as date)")
        self._t(datetime(2011, 2, 3, 10, 11, 12, 3000, tzoffset('', 3 * 60 * 60)), "cast('2011-02-03T10:11:12.003000+03:00' as datetimeoffset)")

    def test_regular(self):
        if not IS_TDS71_PLUS(self.conn):
            self.skipTest('Requires TDS7.1+')
        self._t(None, "cast(NULL as varchar)")
        self._t('test', "cast('test' as varchar)")
        self._t('test ', "cast('test' as char(5))")
        self._t('test', "cast(N'test' as nvarchar)")
        self._t('test ', "cast(N'test' as nchar(5))")
        self._t(Decimal('100.55555'), "cast(100.55555 as decimal(8,5))")
        self._t(Decimal('100.55555'), "cast(100.55555 as numeric(8,5))")
        self._t(b'test', "cast('test' as varbinary)")
        self._t(b'test\x00', "cast('test' as binary(5))")
        self._t(datetime(2011, 2, 3, 10, 11, 12, 3000), "cast('2011-02-03T10:11:12.003' as datetime)")
        self._t(datetime(2011, 2, 3, 10, 11, 0), "cast('2011-02-03T10:11:00' as smalldatetime)")
        val = uuid.uuid4()
        self._t(val, "cast('{}' as uniqueidentifier)".format(val))
        self._t(True, "cast(1 as bit)")
        self._t(128, "cast(128 as tinyint)")
        self._t(255, "cast(255 as tinyint)")
        self._t(-32000, "cast(-32000 as smallint)")
        self._t(2000000000, "cast(2000000000 as int)")
        self._t(2000000000000, "cast(2000000000000 as bigint)")
        self._t(0.12345, "cast(0.12345 as float)")
        self._t(0.25, "cast(0.25 as real)")
        self._t(Decimal('922337203685477.5807'), "cast('922,337,203,685,477.5807' as money)")
        self._t(Decimal('-214748.3648'), "cast('- 214,748.3648' as smallmoney)")


class BadConnection(unittest.TestCase):
    def test_invalid_parameters(self):
        with self.assertRaises(Error):
            with connect(server=settings.HOST, database='master', user=settings.USER, password=settings.PASSWORD + 'bad') as conn:
                with conn.cursor() as cur:
                    cur.execute('select 1')
        with self.assertRaises(Error):
            with connect(server=settings.HOST + 'bad', database='master', user=settings.USER + 'bad', password=settings.PASSWORD) as conn:
                with conn.cursor() as cur:
                    cur.execute('select 1')
        with self.assertRaises(Error):
            with connect(server=settings.HOST, database='doesnotexist', user=settings.USER, password=settings.PASSWORD) as conn:
                with conn.cursor() as cur:
                    cur.execute('select 1')
        with self.assertRaises(Error):
            with connect(server=settings.HOST, database='master', user=settings.USER, password=None) as conn:
                with conn.cursor() as cur:
                    cur.execute('select 1')

    def test_bad_instance_name(self):
        if not hasattr(settings, 'INSTANCE_PORT'):
            return self.skipTest('INSTANCE_PORT must be set to run this test')
        with self.assertRaisesRegexp(LoginError, 'Invalid instance name'):
            host = settings.HOST
            if '\\' in host:
                host, _ = host.split('\\')
            with connect(server=host + '\\badinstancename', database='master', user=settings.USER, password=settings.PASSWORD, port=settings.INSTANCE_PORT) as conn:
                with conn.cursor() as cur:
                    cur.execute('select 1')


def get_spid(conn):
    with conn.cursor() as cur:
        return cur.spid


def kill(conn, spid):
    with conn.cursor() as cur:
        cur.execute('kill {}'.format(spid))


class ConnectionClosing(unittest.TestCase):
    def test_open_close(self):
        for x in xrange(3):
            connect(server=settings.HOST, database='master', user=settings.USER, password=settings.PASSWORD).close()

    def test_closing_after_closed_by_server(self):
        '''
        You should be able to call close on connection closed by server
        '''
        with connect(server=settings.HOST,
                     database='master',
                     user=settings.USER,
                     password=settings.PASSWORD,
                     autocommit=True) as master_conn:
            with connect(server=settings.HOST,
                         database='master',
                         user=settings.USER,
                         password=settings.PASSWORD,
                         autocommit=False,
                         use_mars=settings.USE_MARS) as conn:
                with conn.cursor() as cur:
                    cur.execute('select 1')
                    conn.commit()
                    kill(master_conn, get_spid(conn))
                    sleep(0.2)
                conn.close()

    def test_connection_closed_by_server(self):
        with connect(server=settings.HOST,
                     database='master',
                     user=settings.USER,
                     password=settings.PASSWORD,
                     autocommit=True) as master_conn:
            with connect(server=settings.HOST,
                         database='master',
                         user=settings.USER,
                         password=settings.PASSWORD,
                         autocommit=False,
                         use_mars=settings.USE_MARS) as conn:
                # test overall recovery
                with conn.cursor() as cur:
                    cur.execute('select 1')
                    conn.commit()
                    kill(master_conn, get_spid(conn))
                    sleep(0.2)
                    cur.execute('select 1')
                    cur.fetchall()
                kill(master_conn, get_spid(conn))
                sleep(0.2)
                with conn.cursor() as cur:
                    cur.execute('select 1')

                # test recovery on transaction
                with conn.cursor() as cur:
                    cur.execute('create table ##testtable3 (fld int)')
                    kill(master_conn, get_spid(conn))
                    sleep(0.2)
                    with self.assertRaises(Exception):
                        cur.execute('select * from ##testtable2')
                        cur.fetchall()
                    conn.rollback()
                    cur.execute('select 1')
            #with connect(server=settings.HOST, database='master', user=settings.USER, password=settings.PASSWORD) as conn:
            #    spid = get_spid(conn)
            #    with conn.cursor() as cur:
            #        # test recovery of specific lowlevel methods
            #        tds_submit_query(cur._session, "waitfor delay '00:00:05'; select 1")
            #        kill(master_conn, spid)
            #        self.assertTrue(cur._session.is_connected())
            #        with self.assertRaises(Exception):
            #            tds_process_tokens(cur._session, TDS_TOKEN_RESULTS)
            #        self.assertFalse(cur._session.is_connected())


class Bug1(TestCase):
    def runTest(self):
        try:
            with connect(server=settings.HOST, database='master', user=settings.USER, password=settings.PASSWORD + '<PASSWORD>') as conn:
                with conn.cursor() as cur:
                    cur.execute('select 1')
                    cur.fetchall()
                conn.rollback()
        except:
            pass


#class EncryptionTest(unittest.TestCase):
#    def runTest(self):
#        conn = connect(server=settings.HOST, database='master', user=settings.USER, password=<PASSWORD>, encryption_level=TDS_ENCRYPTION_REQUIRE)
#        cur = conn.cursor()
#        cur.execute('select 1')


class Bug2(DbTestCase):
    def runTest(self):
        with self.conn.cursor() as cur:
            cur.execute('''
            create procedure testproc (@param int)
            as
            begin
                set transaction isolation level read uncommitted -- that will produce very empty result (even no rowcount)
                select @param
                return @param + 1
            end
            ''')
            val = 45
            cur.execute('exec testproc @param = 45')
            self.assertEqual(cur.fetchall(), [(val,)])
            self.assertEqual(val + 1, cur.get_proc_return_status())


class Bug3(TestCase):
    def runTest(self):
        with self.conn.cursor() as cur:
            cur.close()


class DateAndTimeParams(TestCase):
    def test_date(self):
        if not IS_TDS73_PLUS(self.conn):
            self.skipTest('Requires TDS7.3+')
        with self.conn.cursor() as cur:
            date = Date(2012, 10, 6)
            cur.execute('select %s', (date, ))
            self.assertEqual(cur.fetchall(), [(date,)])

    def test_time(self):
        if not IS_TDS73_PLUS(self.conn):
            self.skipTest('Requires TDS7.3+')
        with self.conn.cursor() as cur:
            time = Time(8, 7, 4, 123000)
            cur.execute('select %s', (time, ))
            self.assertEqual(cur.fetchall(), [(time,)])

    def test_datetime(self):
        with self.conn.cursor() as cur:
            time = Timestamp(2013, 7, 9, 8, 7, 4, 123000)
            cur.execute('select %s', (time, ))
            self.assertEqual(cur.fetchall(), [(time,)])


class Extensions(TestCase):
    def runTest(self):
        with self.conn.cursor() as cur:
            self.assertEqual(cur.connection, self.conn)


class SmallDateTimeTest(TestCase):
    def _testval(self, val):
        with self.conn.cursor() as cur:
            cur.execute('select cast(%s as smalldatetime)', (val,))
            self.assertEqual(cur.fetchall(), [(val,)])

    def runTest(self):
        self._testval(Timestamp(2010, 2, 1, 10, 12, 0))
        self._testval(Timestamp(1900, 1, 1, 0, 0, 0))
        self._testval(Timestamp(2079, 6, 6, 23, 59, 0))
        with self.assertRaises(Error):
            self._testval(Timestamp(1899, 1, 1, 0, 0, 0))
        with self.assertRaises(Error):
            self._testval(Timestamp(2080, 1, 1, 0, 0, 0))


class DateTimeTest(DbTestCase):
    def _testencdec(self, val):
        self.assertEqual(val, DateTime.decode(*DateTime._struct.unpack(DateTime.encode(val))))

    def _testval(self, val):
        with self.conn.cursor() as cur:
            cur.execute('select cast(%s as datetime)', (val,))
            self.assertEqual(cur.fetchall(), [(val,)])

    def runTest(self):
        self.assertEqual(DateTime.decode(*DateTime._struct.unpack(b'\xf2\x9c\x00\x00}uO\x01')), Timestamp(2010, 1, 2, 20, 21, 22, 123000))
        self.assertEqual(DateTime.decode(*DateTime._struct.unpack(b'\x7f$-\x00\xff\x81\x8b\x01')), DateTime._max_date)
        self.assertEqual(b'\xf2\x9c\x00\x00}uO\x01', DateTime.encode(Timestamp(2010, 1, 2, 20, 21, 22, 123000)))
        self.assertEqual(b'\x7f$-\x00\xff\x81\x8b\x01', DateTime.encode(DateTime._max_date))
        with self.conn.cursor() as cur:
            cur.execute("select cast('9999-12-31T23:59:59.997' as datetime)")
            self.assertEqual(cur.fetchall(), [(Timestamp(9999, 12, 31, 23, 59, 59, 997000),)])
        self._testencdec(Timestamp(2010, 1, 2, 10, 11, 12))
        self._testval(Timestamp(2010, 1, 2, 0, 0, 0))
        self._testval(Timestamp(2010, 1, 2, 10, 11, 12))
        self._testval(Timestamp(1753, 1, 1, 0, 0, 0))
        self._testval(Timestamp(9999, 12, 31, 0, 0, 0))
        with self.conn.cursor() as cur:
            cur.execute("select cast(null as datetime)")
            self.assertEqual(cur.fetchall(), [(None,)])
        self._testval(Timestamp(9999, 12, 31, 23, 59, 59, 997000))
        with self.assertRaises(Error):
            self._testval(Timestamp(1752, 1, 1, 0, 0, 0))
        with self.conn.cursor() as cur:
            cur.execute('''
            if object_id('testtable') is not null
                drop table testtable
            ''')
            cur.execute('create table testtable (col datetime not null)')
            dt = Timestamp(2010, 1, 2, 20, 21, 22, 123000)
            cur.execute('insert into testtable values (%s)', (dt,))
            cur.execute('select col from testtable')
            self.assertEqual(cur.fetchone(), (dt,))


class NewDateTimeTest(TestCase):
    def test_datetimeoffset(self):
        if not IS_TDS73_PLUS(self.conn):
            self.skipTest('Requires TDS7.3+')

        def _testval(val):
            with self.conn.cursor() as cur:
                cur.execute('select cast(%s as datetimeoffset)', (val,))
                self.assertEqual(cur.fetchall(), [(val,)])

        with self.conn.cursor() as cur:
            cur.execute("select cast('2010-01-02T20:21:22.1234567+05:00' as datetimeoffset)")
            self.assertEqual(datetime(2010, 1, 2, 20, 21, 22, 123456, tzoffset('', 5 * 60 * 60)), cur.fetchone()[0])
        _testval(Timestamp(2010, 1, 2, 0, 0, 0, 0, tzutc()))
        _testval(Timestamp(2010, 1, 2, 0, 0, 0, 0, tzoffset('', 5 * 60 * 60)))
        _testval(Timestamp(1, 1, 1, 0, 0, 0, 0, tzutc()))
        _testval(Timestamp(9999, 12, 31, 23, 59, 59, 999999, tzutc()))
        _testval(Timestamp(2010, 1, 2, 0, 0, 0, 0, tzoffset('', 14 * 60)))
        _testval(Timestamp(2010, 1, 2, 0, 0, 0, 0, tzoffset('', -14 * 60)))
        _testval(Timestamp(2010, 1, 2, 0, 0, 0, 0, tzoffset('', -15 * 60)))

    def test_time(self):
        if not IS_TDS73_PLUS(self.conn):
            self.skipTest('Requires TDS7.3+')

        def testval(val):
            with self.conn.cursor() as cur:
                cur.execute('select cast(%s as time)', (val,))
                self.assertEqual(cur.fetchall(), [(val,)])

        testval(Time(14, 16, 18, 123456))
        testval(Time(0, 0, 0, 0))
        testval(Time(0, 0, 0, 0))
        testval(Time(0, 0, 0, 0))
        testval(Time(23, 59, 59, 999999))
        testval(Time(0, 0, 0, 0))
        testval(Time(0, 0, 0, 0))
        testval(Time(0, 0, 0, 0))

    def test_datetime2(self):
        if not IS_TDS73_PLUS(self.conn):
            self.skipTest('Requires TDS7.3+')

        def testval(val):
            with self.conn.cursor() as cur:
                cur.execute('select cast(%s as datetime2)', (val,))
                self.assertEqual(cur.fetchall(), [(val,)])

        testval(Timestamp(2010, 1, 2, 20, 21, 22, 345678))
        testval(Timestamp(2010, 1, 2, 0, 0, 0))
        testval(Timestamp(1, 1, 1, 0, 0, 0))
        testval(Timestamp(9999, 12, 31, 23, 59, 59, 999999))

    def test_date(self):
        if not IS_TDS73_PLUS(self.conn):
            self.skipTest('Requires TDS7.3+')

        def testval(val):
            with self.conn.cursor() as cur:
                cur.execute('select cast(%s as date)', (val,))
                self.assertEqual(cur.fetchall(), [(val,)])

        testval(Date(2010, 1, 2))
        testval(Date(2010, 1, 2))
        testval(Date(1, 1, 1))
        testval(Date(9999, 12, 31))


class Auth(unittest.TestCase):
    #def test_ntlm(self):
    #    conn = connect(settings.HOST, auth=NtlmAuth(user_name=settings.NTLM_USER, password=settings.NTLM_PASSWORD))
    #    with conn.cursor() as cursor:
    #        cursor.execute('select 1')
    #        cursor.fetchall()

    @unittest.skipUnless(sys.platform.startswith("win"), "requires Windows")
    def test_sspi(self):
        from pytds.login import SspiAuth
        with connect(settings.HOST, auth=SspiAuth()) as conn:
            with conn.cursor() as cursor:
                cursor.execute('select 1')
                cursor.fetchall()

    @unittest.skipIf(getattr(settings, 'SKIP_SQL_AUTH', False), 'SKIP_SQL_AUTH is set')
    def test_sqlauth(self):
        with connect(settings.HOST, user=settings.USER, password=settings.PASSWORD) as conn:
            with conn.cursor() as cursor:
                cursor.execute('select 1')
                cursor.fetchall()


class CloseCursorTwice(TestCase):
    def runTest(self):
        cursor = self.conn.cursor()
        cursor.close()
        cursor.close()


class RegressionSuite(TestCase):
    def test_commit(self):
        if self.conn.mars_enabled:
            self.skipTest('Only breaks when mars is disabled')
        cursor = self.conn.cursor()
        cursor.execute('select 1')
        cursor.fetchall()
        self.conn.commit()

{'fraction_non_alphanumeric': 0.1445705528285543, 'fraction_numerical': 0.051862381449481375, 'mean_word_length': 2.525034253278528, 'pattern_counts': {'":': 0, '<': 22, '<?xml version=': 0, '>': 22, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 14}, 'pii_count': 0, 'substrings_counts': 8, 'word_list_counts': {'cursed_substrings.json': 28, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '295773', 'n_tokens_mistral': 32381, 'n_tokens_neox': 28578, 'n_words': 5844}