installer_files/env/Lib/site-packages/adodbapi/test/adodbapitest.py

""" Unit tests version 2.6.1.0 for adodbapi"""
"""
    adodbapi - A python DB API 2.0 interface to Microsoft ADO

    Copyright (C) 2002  Henrik Ekelund

    This library is free software; you can redistribute it and/or
    modify it under the terms of the GNU Lesser General Public
    License as published by the Free Software Foundation; either
    version 2.1 of the License, or (at your option) any later version.

    This library is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
    Lesser General Public License for more details.

    You should have received a copy of the GNU Lesser General Public
    License along with this library; if not, write to the Free Software
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

    Updates by Vernon Cole
"""

import copy
import datetime
import decimal
import random
import string
import sys
import unittest

try:
    import win32com.client

    win32 = True
except ImportError:
    win32 = False

# run the configuration module.
import adodbapitestconfig as config  # will set sys.path to find correct version of adodbapi

# in our code below, all our switches are from config.whatever
import tryconnection

import adodbapi
import adodbapi.apibase as api

try:
    import adodbapi.ado_consts as ado_consts
except ImportError:  # we are doing a shortcut import as a module -- so
    try:
        import ado_consts
    except ImportError:
        from adodbapi import ado_consts


def str2bytes(sval):
    return sval.encode("latin1")


long = int


def randomstring(length):
    return "".join([random.choice(string.ascii_letters) for n in range(32)])


class CommonDBTests(unittest.TestCase):
    "Self contained super-simple tests in easy syntax, should work on everything between mySQL and Oracle"

    def setUp(self):
        self.engine = "unknown"

    def getEngine(self):
        return self.engine

    def getConnection(self):
        raise NotImplementedError  # "This method must be overriden by a subclass"

    def getCursor(self):
        return self.getConnection().cursor()

    def testConnection(self):
        crsr = self.getCursor()
        assert crsr.__class__.__name__ == "Cursor"

    def testErrorHandlerInherits(self):
        if not self.remote:
            conn = self.getConnection()
            mycallable = lambda connection, cursor, errorclass, errorvalue: 1
            conn.errorhandler = mycallable
            crsr = conn.cursor()
            assert (
                crsr.errorhandler == mycallable
            ), "Error handler on crsr should be same as on connection"

    def testDefaultErrorHandlerConnection(self):
        if not self.remote:
            conn = self.getConnection()
            del conn.messages[:]
            try:
                conn.close()
                conn.commit()  # Should not be able to use connection after it is closed
            except:
                assert len(conn.messages) == 1
                assert len(conn.messages[0]) == 2
                assert conn.messages[0][0] == api.ProgrammingError

    def testOwnErrorHandlerConnection(self):
        if self.remote:  # ToDo: use "skip"
            return
        mycallable = (
            lambda connection, cursor, errorclass, errorvalue: 1
        )  # does not raise anything
        conn = self.getConnection()
        conn.errorhandler = mycallable
        conn.close()
        conn.commit()  # Should not be able to use connection after it is closed
        assert len(conn.messages) == 0

        conn.errorhandler = None  # This should bring back the standard error handler
        try:
            conn.close()
            conn.commit()  # Should not be able to use connection after it is closed
        except:
            pass
        # The Standard errorhandler appends error to messages attribute
        assert (
            len(conn.messages) > 0
        ), "Setting errorhandler to none  should bring back the standard error handler"

    def testDefaultErrorHandlerCursor(self):
        crsr = self.getConnection().cursor()
        if not self.remote:
            del crsr.messages[:]
            try:
                crsr.execute("SELECT abbtytddrf FROM dasdasd")
            except:
                assert len(crsr.messages) == 1
                assert len(crsr.messages[0]) == 2
                assert crsr.messages[0][0] == api.DatabaseError

    def testOwnErrorHandlerCursor(self):
        if self.remote:  # ToDo: should be a "skip"
            return
        mycallable = (
            lambda connection, cursor, errorclass, errorvalue: 1
        )  # does not raise anything
        crsr = self.getConnection().cursor()
        crsr.errorhandler = mycallable
        crsr.execute("SELECT abbtytddrf FROM dasdasd")
        assert len(crsr.messages) == 0

        crsr.errorhandler = None  # This should bring back the standard error handler
        try:
            crsr.execute("SELECT abbtytddrf FROM dasdasd")
        except:
            pass
        # The Standard errorhandler appends error to messages attribute
        assert (
            len(crsr.messages) > 0
        ), "Setting errorhandler to none  should bring back the standard error handler"

    def testUserDefinedConversions(self):
        if self.remote:  ## Todo: should be a "skip"
            return
        try:
            duplicatingConverter = lambda aStringField: aStringField * 2
            assert duplicatingConverter("gabba") == "gabbagabba"

            self.helpForceDropOnTblTemp()
            conn = self.getConnection()
            # the variantConversions attribute should not exist on a normal connection object
            self.assertRaises(AttributeError, lambda x: conn.variantConversions[x], [2])
            if not self.remote:
                # create a variantConversions attribute on the connection
                conn.variantConversions = copy.copy(api.variantConversions)
                crsr = conn.cursor()
                tabdef = (
                    "CREATE TABLE xx_%s (fldData VARCHAR(100) NOT NULL, fld2 VARCHAR(20))"
                    % config.tmp
                )
                crsr.execute(tabdef)
                crsr.execute(
                    "INSERT INTO xx_%s(fldData,fld2) VALUES('gabba','booga')"
                    % config.tmp
                )
                crsr.execute(
                    "INSERT INTO xx_%s(fldData,fld2) VALUES('hey','yo')" % config.tmp
                )
                # change converter for ALL adoStringTypes columns
                conn.variantConversions[api.adoStringTypes] = duplicatingConverter
                crsr.execute(
                    "SELECT fldData,fld2 FROM xx_%s ORDER BY fldData" % config.tmp
                )

                rows = crsr.fetchall()
                row = rows[0]
                self.assertEqual(row[0], "gabbagabba")
                row = rows[1]
                self.assertEqual(row[0], "heyhey")
                self.assertEqual(row[1], "yoyo")

                upcaseConverter = lambda aStringField: aStringField.upper()
                assert upcaseConverter("upThis") == "UPTHIS"

                # now use a single column converter
                rows.converters[1] = upcaseConverter  # convert second column
                self.assertEqual(row[0], "heyhey")  # first will be unchanged
                self.assertEqual(row[1], "YO")  # second will convert to upper case

        finally:
            try:
                del conn.variantConversions  # Restore the default
            except:
                pass
            self.helpRollbackTblTemp()

    def testUserDefinedConversionForExactNumericTypes(self):
        # variantConversions is a dictionary of conversion functions
        # held internally in adodbapi.apibase
        #
        # !!! this test intentionally alters the value of what should be constant in the module
        # !!! no new code should use this example, to is only a test to see that the
        # !!! deprecated way of doing this still works.  (use connection.variantConversions)
        #
        if not self.remote and sys.version_info < (3, 0):  ### Py3 need different test
            oldconverter = adodbapi.variantConversions[
                ado_consts.adNumeric
            ]  # keep old function to restore later
            # By default decimal and "numbers" are returned as decimals.
            # Instead, make numbers return as  floats
            try:
                adodbapi.variantConversions[ado_consts.adNumeric] = adodbapi.cvtFloat
                self.helpTestDataType(
                    "decimal(18,2)", "NUMBER", 3.45, compareAlmostEqual=1
                )
                self.helpTestDataType(
                    "numeric(18,2)", "NUMBER", 3.45, compareAlmostEqual=1
                )
                # now return strings
                adodbapi.variantConversions[ado_consts.adNumeric] = adodbapi.cvtString
                self.helpTestDataType("numeric(18,2)", "NUMBER", "3.45")
                # now a completly weird user defined convertion
                adodbapi.variantConversions[ado_consts.adNumeric] = (
                    lambda x: "!!This function returns a funny unicode string %s!!" % x
                )
                self.helpTestDataType(
                    "numeric(18,2)",
                    "NUMBER",
                    "3.45",
                    allowedReturnValues=[
                        "!!This function returns a funny unicode string 3.45!!"
                    ],
                )
            finally:
                # now reset the converter to its original function
                adodbapi.variantConversions[
                    ado_consts.adNumeric
                ] = oldconverter  # Restore the original convertion function

    def helpTestDataType(
        self,
        sqlDataTypeString,
        DBAPIDataTypeString,
        pyData,
        pyDataInputAlternatives=None,
        compareAlmostEqual=None,
        allowedReturnValues=None,
    ):
        self.helpForceDropOnTblTemp()
        conn = self.getConnection()
        crsr = conn.cursor()
        tabdef = (
            """
            CREATE TABLE xx_%s (
                fldId integer NOT NULL,
                fldData """
            % config.tmp
            + sqlDataTypeString
            + ")\n"
        )

        crsr.execute(tabdef)

        # Test Null values mapped to None
        crsr.execute("INSERT INTO xx_%s (fldId) VALUES (1)" % config.tmp)

        crsr.execute("SELECT fldId,fldData FROM xx_%s" % config.tmp)
        rs = crsr.fetchone()
        self.assertEqual(rs[1], None)  # Null should be mapped to None
        assert rs[0] == 1

        # Test description related
        descTuple = crsr.description[1]
        assert descTuple[0] in ["fldData", "flddata"], 'was "%s" expected "%s"' % (
            descTuple[0],
            "fldData",
        )

        if DBAPIDataTypeString == "STRING":
            assert descTuple[1] == api.STRING, 'was "%s" expected "%s"' % (
                descTuple[1],
                api.STRING.values,
            )
        elif DBAPIDataTypeString == "NUMBER":
            assert descTuple[1] == api.NUMBER, 'was "%s" expected "%s"' % (
                descTuple[1],
                api.NUMBER.values,
            )
        elif DBAPIDataTypeString == "BINARY":
            assert descTuple[1] == api.BINARY, 'was "%s" expected "%s"' % (
                descTuple[1],
                api.BINARY.values,
            )
        elif DBAPIDataTypeString == "DATETIME":
            assert descTuple[1] == api.DATETIME, 'was "%s" expected "%s"' % (
                descTuple[1],
                api.DATETIME.values,
            )
        elif DBAPIDataTypeString == "ROWID":
            assert descTuple[1] == api.ROWID, 'was "%s" expected "%s"' % (
                descTuple[1],
                api.ROWID.values,
            )
        elif DBAPIDataTypeString == "UUID":
            assert descTuple[1] == api.OTHER, 'was "%s" expected "%s"' % (
                descTuple[1],
                api.OTHER.values,
            )
        else:
            raise NotImplementedError  # "DBAPIDataTypeString not provided"

        # Test data binding
        inputs = [pyData]
        if pyDataInputAlternatives:
            inputs.extend(pyDataInputAlternatives)
        inputs = set(inputs)  # removes redundant string==unicode tests
        fldId = 1
        for inParam in inputs:
            fldId += 1
            try:
                crsr.execute(
                    "INSERT INTO xx_%s (fldId,fldData) VALUES (?,?)" % config.tmp,
                    (fldId, inParam),
                )
            except:
                if self.remote:
                    for message in crsr.messages:
                        print(message)
                else:
                    conn.printADOerrors()
                raise
            crsr.execute(
                "SELECT fldData FROM xx_%s WHERE ?=fldID" % config.tmp, [fldId]
            )
            rs = crsr.fetchone()
            if allowedReturnValues:
                allowedTypes = tuple([type(aRV) for aRV in allowedReturnValues])
                assert isinstance(
                    rs[0], allowedTypes
                ), 'result type "%s" must be one of %s' % (type(rs[0]), allowedTypes)
            else:
                assert isinstance(
                    rs[0], type(pyData)
                ), 'result type "%s" must be instance of %s' % (
                    type(rs[0]),
                    type(pyData),
                )

            if compareAlmostEqual and DBAPIDataTypeString == "DATETIME":
                iso1 = adodbapi.dateconverter.DateObjectToIsoFormatString(rs[0])
                iso2 = adodbapi.dateconverter.DateObjectToIsoFormatString(pyData)
                self.assertEqual(iso1, iso2)
            elif compareAlmostEqual:
                s = float(pyData)
                v = float(rs[0])
                assert (
                    abs(v - s) / s < 0.00001
                ), "Values not almost equal recvd=%s, expected=%f" % (rs[0], s)
            else:
                if allowedReturnValues:
                    ok = False
                    self.assertTrue(
                        rs[0] in allowedReturnValues,
                        'Value "%s" not in %s' % (repr(rs[0]), allowedReturnValues),
                    )
                else:
                    self.assertEqual(
                        rs[0],
                        pyData,
                        'Values are not equal recvd="%s", expected="%s"'
                        % (rs[0], pyData),
                    )

    def testDataTypeFloat(self):
        self.helpTestDataType("real", "NUMBER", 3.45, compareAlmostEqual=True)
        self.helpTestDataType("float", "NUMBER", 1.79e37, compareAlmostEqual=True)

    def testDataTypeDecmal(self):
        self.helpTestDataType(
            "decimal(18,2)",
            "NUMBER",
            3.45,
            allowedReturnValues=["3.45", "3,45", decimal.Decimal("3.45")],
        )
        self.helpTestDataType(
            "numeric(18,2)",
            "NUMBER",
            3.45,
            allowedReturnValues=["3.45", "3,45", decimal.Decimal("3.45")],
        )
        self.helpTestDataType(
            "decimal(20,2)",
            "NUMBER",
            444444444444444444,
            allowedReturnValues=[
                "444444444444444444.00",
                "444444444444444444,00",
                decimal.Decimal("444444444444444444"),
            ],
        )
        if self.getEngine() == "MSSQL":
            self.helpTestDataType(
                "uniqueidentifier",
                "UUID",
                "{71A4F49E-39F3-42B1-A41E-48FF154996E6}",
                allowedReturnValues=["{71A4F49E-39F3-42B1-A41E-48FF154996E6}"],
            )

    def testDataTypeMoney(self):  # v2.1 Cole -- use decimal for money
        if self.getEngine() == "MySQL":
            self.helpTestDataType(
                "DECIMAL(20,4)", "NUMBER", decimal.Decimal("-922337203685477.5808")
            )
        elif self.getEngine() == "PostgreSQL":
            self.helpTestDataType(
                "money",
                "NUMBER",
                decimal.Decimal("-922337203685477.5808"),
                compareAlmostEqual=True,
                allowedReturnValues=[
                    -922337203685477.5808,
                    decimal.Decimal("-922337203685477.5808"),
                ],
            )
        else:
            self.helpTestDataType("smallmoney", "NUMBER", decimal.Decimal("214748.02"))
            self.helpTestDataType(
                "money", "NUMBER", decimal.Decimal("-922337203685477.5808")
            )

    def testDataTypeInt(self):
        if self.getEngine() != "PostgreSQL":
            self.helpTestDataType("tinyint", "NUMBER", 115)
        self.helpTestDataType("smallint", "NUMBER", -32768)
        if self.getEngine() not in ["ACCESS", "PostgreSQL"]:
            self.helpTestDataType(
                "bit", "NUMBER", 1
            )  # Does not work correctly with access
        if self.getEngine() in ["MSSQL", "PostgreSQL"]:
            self.helpTestDataType(
                "bigint",
                "NUMBER",
                3000000000,
                allowedReturnValues=[3000000000, int(3000000000)],
            )
        self.helpTestDataType("int", "NUMBER", 2147483647)

    def testDataTypeChar(self):
        for sqlDataType in ("char(6)", "nchar(6)"):
            self.helpTestDataType(
                sqlDataType,
                "STRING",
                "spam  ",
                allowedReturnValues=["spam", "spam", "spam  ", "spam  "],
            )

    def testDataTypeVarChar(self):
        if self.getEngine() == "MySQL":
            stringKinds = ["varchar(10)", "text"]
        elif self.getEngine() == "PostgreSQL":
            stringKinds = ["varchar(10)", "text", "character varying"]
        else:
            stringKinds = [
                "varchar(10)",
                "nvarchar(10)",
                "text",
                "ntext",
            ]  # ,"varchar(max)"]

        for sqlDataType in stringKinds:
            self.helpTestDataType(sqlDataType, "STRING", "spam", ["spam"])

    def testDataTypeDate(self):
        if self.getEngine() == "PostgreSQL":
            dt = "timestamp"
        else:
            dt = "datetime"
        self.helpTestDataType(
            dt, "DATETIME", adodbapi.Date(2002, 10, 28), compareAlmostEqual=True
        )
        if self.getEngine() not in ["MySQL", "PostgreSQL"]:
            self.helpTestDataType(
                "smalldatetime",
                "DATETIME",
                adodbapi.Date(2002, 10, 28),
                compareAlmostEqual=True,
            )
        if tag != "pythontime" and self.getEngine() not in [
            "MySQL",
            "PostgreSQL",
        ]:  # fails when using pythonTime
            self.helpTestDataType(
                dt,
                "DATETIME",
                adodbapi.Timestamp(2002, 10, 28, 12, 15, 1),
                compareAlmostEqual=True,
            )

    def testDataTypeBinary(self):
        binfld = str2bytes("\x07\x00\xE2\x40*")
        arv = [binfld, adodbapi.Binary(binfld), bytes(binfld)]
        if self.getEngine() == "PostgreSQL":
            self.helpTestDataType(
                "bytea", "BINARY", adodbapi.Binary(binfld), allowedReturnValues=arv
            )
        else:
            self.helpTestDataType(
                "binary(5)", "BINARY", adodbapi.Binary(binfld), allowedReturnValues=arv
            )
            self.helpTestDataType(
                "varbinary(100)",
                "BINARY",
                adodbapi.Binary(binfld),
                allowedReturnValues=arv,
            )
            if self.getEngine() != "MySQL":
                self.helpTestDataType(
                    "image", "BINARY", adodbapi.Binary(binfld), allowedReturnValues=arv
                )

    def helpRollbackTblTemp(self):
        self.helpForceDropOnTblTemp()

    def helpForceDropOnTblTemp(self):
        conn = self.getConnection()
        with conn.cursor() as crsr:
            try:
                crsr.execute("DROP TABLE xx_%s" % config.tmp)
                if not conn.autocommit:
                    conn.commit()
            except:
                pass

    def helpCreateAndPopulateTableTemp(self, crsr):
        tabdef = (
            """
            CREATE TABLE xx_%s (
                fldData INTEGER
            )
            """
            % config.tmp
        )
        try:  # EAFP
            crsr.execute(tabdef)
        except api.DatabaseError:  # was not dropped before
            self.helpForceDropOnTblTemp()  # so drop it now
            crsr.execute(tabdef)
        for i in range(9):  # note: this poor SQL code, but a valid test
            crsr.execute("INSERT INTO xx_%s (fldData) VALUES (%i)" % (config.tmp, i))
            # NOTE: building the test table without using parameter substitution

    def testFetchAll(self):
        crsr = self.getCursor()
        self.helpCreateAndPopulateTableTemp(crsr)
        crsr.execute("SELECT fldData FROM xx_%s" % config.tmp)
        rs = crsr.fetchall()
        assert len(rs) == 9
        # test slice of rows
        i = 3
        for row in rs[3:-2]:  # should have rowid 3..6
            assert row[0] == i
            i += 1
        self.helpRollbackTblTemp()

    def testPreparedStatement(self):
        crsr = self.getCursor()
        self.helpCreateAndPopulateTableTemp(crsr)
        crsr.prepare("SELECT fldData FROM xx_%s" % config.tmp)
        crsr.execute(crsr.command)  # remember the one that was prepared
        rs = crsr.fetchall()
        assert len(rs) == 9
        assert rs[2][0] == 2
        self.helpRollbackTblTemp()

    def testWrongPreparedStatement(self):
        crsr = self.getCursor()
        self.helpCreateAndPopulateTableTemp(crsr)
        crsr.prepare("SELECT * FROM nowhere")
        crsr.execute(
            "SELECT fldData FROM xx_%s" % config.tmp
        )  # should execute this one, not the prepared one
        rs = crsr.fetchall()
        assert len(rs) == 9
        assert rs[2][0] == 2
        self.helpRollbackTblTemp()

    def testIterator(self):
        crsr = self.getCursor()
        self.helpCreateAndPopulateTableTemp(crsr)
        crsr.execute("SELECT fldData FROM xx_%s" % config.tmp)
        for i, row in enumerate(
            crsr
        ):  # using cursor as an iterator, rather than fetchxxx
            assert row[0] == i
        self.helpRollbackTblTemp()

    def testExecuteMany(self):
        crsr = self.getCursor()
        self.helpCreateAndPopulateTableTemp(crsr)
        seq_of_values = [(111,), (222,)]
        crsr.executemany(
            "INSERT INTO xx_%s (fldData) VALUES (?)" % config.tmp, seq_of_values
        )
        if crsr.rowcount == -1:
            print(
                self.getEngine()
                + " Provider does not support rowcount (on .executemany())"
            )
        else:
            self.assertEqual(crsr.rowcount, 2)
        crsr.execute("SELECT fldData FROM xx_%s" % config.tmp)
        rs = crsr.fetchall()
        assert len(rs) == 11
        self.helpRollbackTblTemp()

    def testRowCount(self):
        crsr = self.getCursor()
        self.helpCreateAndPopulateTableTemp(crsr)
        crsr.execute("SELECT fldData FROM xx_%s" % config.tmp)
        if crsr.rowcount == -1:
            # print("provider does not support rowcount on select")
            pass
        else:
            self.assertEqual(crsr.rowcount, 9)
        self.helpRollbackTblTemp()

    def testRowCountNoRecordset(self):
        crsr = self.getCursor()
        self.helpCreateAndPopulateTableTemp(crsr)
        crsr.execute("DELETE FROM xx_%s WHERE fldData >= 5" % config.tmp)
        if crsr.rowcount == -1:
            print(self.getEngine() + " Provider does not support rowcount (on DELETE)")
        else:
            self.assertEqual(crsr.rowcount, 4)
        self.helpRollbackTblTemp()

    def testFetchMany(self):
        crsr = self.getCursor()
        self.helpCreateAndPopulateTableTemp(crsr)
        crsr.execute("SELECT fldData FROM xx_%s" % config.tmp)
        rs = crsr.fetchmany(3)
        assert len(rs) == 3
        rs = crsr.fetchmany(5)
        assert len(rs) == 5
        rs = crsr.fetchmany(5)
        assert len(rs) == 1  # Asked for five, but there is only one left
        self.helpRollbackTblTemp()

    def testFetchManyWithArraySize(self):
        crsr = self.getCursor()
        self.helpCreateAndPopulateTableTemp(crsr)
        crsr.execute("SELECT fldData FROM xx_%s" % config.tmp)
        rs = crsr.fetchmany()
        assert len(rs) == 1  # arraysize Defaults to one
        crsr.arraysize = 4
        rs = crsr.fetchmany()
        assert len(rs) == 4
        rs = crsr.fetchmany()
        assert len(rs) == 4
        rs = crsr.fetchmany()
        assert len(rs) == 0
        self.helpRollbackTblTemp()

    def testErrorConnect(self):
        conn = self.getConnection()
        kw = {}
        if "proxy_host" in conn.kwargs:
            kw["proxy_host"] = conn.kwargs["proxy_host"]
        conn.close()
        self.assertRaises(api.DatabaseError, self.db, "not a valid connect string", kw)

    def testRowIterator(self):
        self.helpForceDropOnTblTemp()
        conn = self.getConnection()
        crsr = conn.cursor()
        tabdef = (
            """
            CREATE TABLE xx_%s (
                fldId integer NOT NULL,
                fldTwo integer,
                fldThree integer,
                fldFour integer)
                """
            % config.tmp
        )
        crsr.execute(tabdef)

        inputs = [(2, 3, 4), (102, 103, 104)]
        fldId = 1
        for inParam in inputs:
            fldId += 1
            try:
                crsr.execute(
                    "INSERT INTO xx_%s (fldId,fldTwo,fldThree,fldFour) VALUES (?,?,?,?)"
                    % config.tmp,
                    (fldId, inParam[0], inParam[1], inParam[2]),
                )
            except:
                if self.remote:
                    for message in crsr.messages:
                        print(message)
                else:
                    conn.printADOerrors()
                raise
            crsr.execute(
                "SELECT fldTwo,fldThree,fldFour FROM xx_%s WHERE ?=fldID" % config.tmp,
                [fldId],
            )
            rec = crsr.fetchone()
            # check that stepping through an emulated row works
            for j in range(len(inParam)):
                assert (
                    rec[j] == inParam[j]
                ), 'returned value:"%s" != test value:"%s"' % (rec[j], inParam[j])
            # check that we can get a complete tuple from a row
            assert tuple(rec) == inParam, 'returned value:"%s" != test value:"%s"' % (
                repr(rec),
                repr(inParam),
            )
            # test that slices of rows work
            slice1 = tuple(rec[:-1])
            slice2 = tuple(inParam[0:2])
            assert slice1 == slice2, 'returned value:"%s" != test value:"%s"' % (
                repr(slice1),
                repr(slice2),
            )
            # now test named column retrieval
            assert rec["fldTwo"] == inParam[0]
            assert rec.fldThree == inParam[1]
            assert rec.fldFour == inParam[2]
        # test array operation
        # note that the fields vv        vv     vv    are out of order
        crsr.execute("select fldThree,fldFour,fldTwo from xx_%s" % config.tmp)
        recs = crsr.fetchall()
        assert recs[1][0] == 103
        assert recs[0][1] == 4
        assert recs[1]["fldFour"] == 104
        assert recs[0, 0] == 3
        assert recs[0, "fldTwo"] == 2
        assert recs[1, 2] == 102
        for i in range(1):
            for j in range(2):
                assert recs[i][j] == recs[i, j]

    def testFormatParamstyle(self):
        self.helpForceDropOnTblTemp()
        conn = self.getConnection()
        conn.paramstyle = "format"  # test nonstandard use of paramstyle
        crsr = conn.cursor()
        tabdef = (
            """
            CREATE TABLE xx_%s (
                fldId integer NOT NULL,
                fldData varchar(10),
                fldConst varchar(30))
                """
            % config.tmp
        )
        crsr.execute(tabdef)

        inputs = ["one", "two", "three"]
        fldId = 2
        for inParam in inputs:
            fldId += 1
            sql = (
                "INSERT INTO xx_"
                + config.tmp
                + " (fldId,fldConst,fldData) VALUES (%s,'thi%s :may cause? trouble', %s)"
            )
            try:
                crsr.execute(sql, (fldId, inParam))
            except:
                if self.remote:
                    for message in crsr.messages:
                        print(message)
                else:
                    conn.printADOerrors()
                raise
            crsr.execute(
                "SELECT fldData, fldConst FROM xx_" + config.tmp + " WHERE %s=fldID",
                [fldId],
            )
            rec = crsr.fetchone()
            self.assertEqual(
                rec[0],
                inParam,
                'returned value:"%s" != test value:"%s"' % (rec[0], inParam),
            )
            self.assertEqual(rec[1], "thi%s :may cause? trouble")

        # now try an operation with a "%s" as part of a literal
        sel = (
            "insert into xx_" + config.tmp + " (fldId,fldData) VALUES (%s,'four%sfive')"
        )
        params = (20,)
        crsr.execute(sel, params)

        # test the .query implementation
        assert "(?," in crsr.query, 'expected:"%s" in "%s"' % ("(?,", crsr.query)
        # test the .command attribute
        assert crsr.command == sel, 'expected:"%s" but found "%s"' % (sel, crsr.command)

        # test the .parameters attribute
        if not self.remote:  # parameter list will be altered in transit
            self.assertEqual(crsr.parameters, params)
        # now make sure the data made it
        crsr.execute("SELECT fldData FROM xx_%s WHERE fldID=20" % config.tmp)
        rec = crsr.fetchone()
        self.assertEqual(rec[0], "four%sfive")

    def testNamedParamstyle(self):
        self.helpForceDropOnTblTemp()
        conn = self.getConnection()
        crsr = conn.cursor()
        crsr.paramstyle = "named"  # test nonstandard use of paramstyle
        tabdef = (
            """
            CREATE TABLE xx_%s (
                fldId integer NOT NULL,
                fldData varchar(10))
                """
            % config.tmp
        )
        crsr.execute(tabdef)

        inputs = ["four", "five", "six"]
        fldId = 10
        for inParam in inputs:
            fldId += 1
            try:
                crsr.execute(
                    "INSERT INTO xx_%s (fldId,fldData) VALUES (:Id,:f_Val)"
                    % config.tmp,
                    {"f_Val": inParam, "Id": fldId},
                )
            except:
                if self.remote:
                    for message in crsr.messages:
                        print(message)
                else:
                    conn.printADOerrors()
                raise
            crsr.execute(
                "SELECT fldData FROM xx_%s WHERE fldID=:Id" % config.tmp, {"Id": fldId}
            )
            rec = crsr.fetchone()
            self.assertEqual(
                rec[0],
                inParam,
                'returned value:"%s" != test value:"%s"' % (rec[0], inParam),
            )
        # now a test with a ":" as part of a literal
        crsr.execute(
            "insert into xx_%s (fldId,fldData) VALUES (:xyz,'six:five')" % config.tmp,
            {"xyz": 30},
        )
        crsr.execute("SELECT fldData FROM xx_%s WHERE fldID=30" % config.tmp)
        rec = crsr.fetchone()
        self.assertEqual(rec[0], "six:five")

    def testPyformatParamstyle(self):
        self.helpForceDropOnTblTemp()
        conn = self.getConnection()
        crsr = conn.cursor()
        crsr.paramstyle = "pyformat"  # test nonstandard use of paramstyle
        tabdef = (
            """
            CREATE TABLE xx_%s (
                fldId integer NOT NULL,
                fldData varchar(10))
                """
            % config.tmp
        )
        crsr.execute(tabdef)

        inputs = ["four", "five", "six"]
        fldId = 10
        for inParam in inputs:
            fldId += 1
            try:
                crsr.execute(
                    "INSERT INTO xx_%s (fldId,fldData) VALUES (%%(Id)s,%%(f_Val)s)"
                    % config.tmp,
                    {"f_Val": inParam, "Id": fldId},
                )
            except:
                if self.remote:
                    for message in crsr.messages:
                        print(message)
                else:
                    conn.printADOerrors()
                raise
            crsr.execute(
                "SELECT fldData FROM xx_%s WHERE fldID=%%(Id)s" % config.tmp,
                {"Id": fldId},
            )
            rec = crsr.fetchone()
            self.assertEqual(
                rec[0],
                inParam,
                'returned value:"%s" != test value:"%s"' % (rec[0], inParam),
            )
        # now a test with a "%" as part of a literal
        crsr.execute(
            "insert into xx_%s (fldId,fldData) VALUES (%%(xyz)s,'six%%five')"
            % config.tmp,
            {"xyz": 30},
        )
        crsr.execute("SELECT fldData FROM xx_%s WHERE fldID=30" % config.tmp)
        rec = crsr.fetchone()
        self.assertEqual(rec[0], "six%five")

    def testAutomaticParamstyle(self):
        self.helpForceDropOnTblTemp()
        conn = self.getConnection()
        conn.paramstyle = "dynamic"  # test nonstandard use of paramstyle
        crsr = conn.cursor()
        tabdef = (
            """
            CREATE TABLE xx_%s (
                fldId integer NOT NULL,
                fldData varchar(10),
                fldConst varchar(30))
                """
            % config.tmp
        )
        crsr.execute(tabdef)
        inputs = ["one", "two", "three"]
        fldId = 2
        for inParam in inputs:
            fldId += 1
            try:
                crsr.execute(
                    "INSERT INTO xx_"
                    + config.tmp
                    + " (fldId,fldConst,fldData) VALUES (?,'thi%s :may cause? troub:1e', ?)",
                    (fldId, inParam),
                )
            except:
                if self.remote:
                    for message in crsr.messages:
                        print(message)
                else:
                    conn.printADOerrors()
                raise
            trouble = "thi%s :may cause? troub:1e"
            crsr.execute(
                "SELECT fldData, fldConst FROM xx_" + config.tmp + " WHERE ?=fldID",
                [fldId],
            )
            rec = crsr.fetchone()
            self.assertEqual(
                rec[0],
                inParam,
                'returned value:"%s" != test value:"%s"' % (rec[0], inParam),
            )
            self.assertEqual(rec[1], trouble)
        #     inputs = [u'four',u'five',u'six']
        fldId = 10
        for inParam in inputs:
            fldId += 1
            try:
                crsr.execute(
                    "INSERT INTO xx_%s (fldId,fldData) VALUES (:Id,:f_Val)"
                    % config.tmp,
                    {"f_Val": inParam, "Id": fldId},
                )
            except:
                if self.remote:
                    for message in crsr.messages:
                        print(message)
                else:
                    conn.printADOerrors()
                raise
            crsr.execute(
                "SELECT fldData FROM xx_%s WHERE :Id=fldID" % config.tmp, {"Id": fldId}
            )
            rec = crsr.fetchone()
            self.assertEqual(
                rec[0],
                inParam,
                'returned value:"%s" != test value:"%s"' % (rec[0], inParam),
            )
        # now a test with a ":" as part of a literal -- and use a prepared query
        ppdcmd = (
            "insert into xx_%s (fldId,fldData) VALUES (:xyz,'six:five')" % config.tmp
        )
        crsr.prepare(ppdcmd)
        crsr.execute(ppdcmd, {"xyz": 30})
        crsr.execute("SELECT fldData FROM xx_%s WHERE fldID=30" % config.tmp)
        rec = crsr.fetchone()
        self.assertEqual(rec[0], "six:five")

    def testRollBack(self):
        conn = self.getConnection()
        crsr = conn.cursor()
        assert not crsr.connection.autocommit, "Unexpected beginning condition"
        self.helpCreateAndPopulateTableTemp(crsr)
        crsr.connection.commit()  # commit the first bunch

        crsr.execute("INSERT INTO xx_%s (fldData) VALUES(100)" % config.tmp)

        selectSql = "SELECT fldData FROM xx_%s WHERE fldData=100" % config.tmp
        crsr.execute(selectSql)
        rs = crsr.fetchall()
        assert len(rs) == 1
        self.conn.rollback()
        crsr.execute(selectSql)
        assert (
            crsr.fetchone() == None
        ), "cursor.fetchone should return None if a query retrieves no rows"
        crsr.execute("SELECT fldData from xx_%s" % config.tmp)
        rs = crsr.fetchall()
        assert len(rs) == 9, "the original records should still be present"
        self.helpRollbackTblTemp()

    def testCommit(self):
        try:
            con2 = self.getAnotherConnection()
        except NotImplementedError:
            return  # should be "SKIP" for ACCESS
        assert not con2.autocommit, "default should be manual commit"
        crsr = con2.cursor()
        self.helpCreateAndPopulateTableTemp(crsr)

        crsr.execute("INSERT INTO xx_%s (fldData) VALUES(100)" % config.tmp)
        con2.commit()

        selectSql = "SELECT fldData FROM xx_%s WHERE fldData=100" % config.tmp
        crsr.execute(selectSql)
        rs = crsr.fetchall()
        assert len(rs) == 1
        crsr.close()
        con2.close()
        conn = self.getConnection()
        crsr = self.getCursor()
        with conn.cursor() as crsr:
            crsr.execute(selectSql)
            rs = crsr.fetchall()
            assert len(rs) == 1
            assert rs[0][0] == 100
        self.helpRollbackTblTemp()

    def testAutoRollback(self):
        try:
            con2 = self.getAnotherConnection()
        except NotImplementedError:
            return  # should be "SKIP" for ACCESS
        assert not con2.autocommit, "unexpected beginning condition"
        crsr = con2.cursor()
        self.helpCreateAndPopulateTableTemp(crsr)
        crsr.execute("INSERT INTO xx_%s (fldData) VALUES(100)" % config.tmp)
        selectSql = "SELECT fldData FROM xx_%s WHERE fldData=100" % config.tmp
        crsr.execute(selectSql)
        rs = crsr.fetchall()
        assert len(rs) == 1
        crsr.close()
        con2.close()
        crsr = self.getCursor()
        try:
            crsr.execute(
                selectSql
            )  # closing the connection should have forced rollback
            row = crsr.fetchone()
        except api.DatabaseError:
            row = None  # if the entire table disappeared the rollback was perfect and the test passed
        assert row == None, (
            "cursor.fetchone should return None if a query retrieves no rows. Got %s"
            % repr(row)
        )
        self.helpRollbackTblTemp()

    def testAutoCommit(self):
        try:
            ac_conn = self.getAnotherConnection({"autocommit": True})
        except NotImplementedError:
            return  # should be "SKIP" for ACCESS
        crsr = ac_conn.cursor()
        self.helpCreateAndPopulateTableTemp(crsr)
        crsr.execute("INSERT INTO xx_%s (fldData) VALUES(100)" % config.tmp)
        crsr.close()
        with self.getCursor() as crsr:
            selectSql = "SELECT fldData from xx_%s" % config.tmp
            crsr.execute(
                selectSql
            )  # closing the connection should _not_ have forced rollback
            rs = crsr.fetchall()
            assert len(rs) == 10, "all records should still be present"
        ac_conn.close()
        self.helpRollbackTblTemp()

    def testSwitchedAutoCommit(self):
        try:
            ac_conn = self.getAnotherConnection()
        except NotImplementedError:
            return  # should be "SKIP" for ACCESS
        ac_conn.autocommit = True
        crsr = ac_conn.cursor()
        self.helpCreateAndPopulateTableTemp(crsr)
        crsr.execute("INSERT INTO xx_%s (fldData) VALUES(100)" % config.tmp)
        crsr.close()
        conn = self.getConnection()
        ac_conn.close()
        with self.getCursor() as crsr:
            selectSql = "SELECT fldData from xx_%s" % config.tmp
            crsr.execute(
                selectSql
            )  # closing the connection should _not_ have forced rollback
            rs = crsr.fetchall()
            assert len(rs) == 10, "all records should still be present"
        self.helpRollbackTblTemp()

    def testExtendedTypeHandling(self):
        class XtendString(str):
            pass

        class XtendInt(int):
            pass

        class XtendFloat(float):
            pass

        xs = XtendString(randomstring(30))
        xi = XtendInt(random.randint(-100, 500))
        xf = XtendFloat(random.random())
        self.helpForceDropOnTblTemp()
        conn = self.getConnection()
        crsr = conn.cursor()
        tabdef = (
            """
            CREATE TABLE xx_%s (
                s VARCHAR(40) NOT NULL,
                i INTEGER NOT NULL,
                f REAL NOT NULL)"""
            % config.tmp
        )
        crsr.execute(tabdef)
        crsr.execute(
            "INSERT INTO xx_%s (s, i, f) VALUES (?, ?, ?)" % config.tmp, (xs, xi, xf)
        )
        crsr.close()
        conn = self.getConnection()
        with self.getCursor() as crsr:
            selectSql = "SELECT s, i, f from xx_%s" % config.tmp
            crsr.execute(
                selectSql
            )  # closing the connection should _not_ have forced rollback
            row = crsr.fetchone()
            self.assertEqual(row.s, xs)
            self.assertEqual(row.i, xi)
            self.assertAlmostEqual(row.f, xf)
        self.helpRollbackTblTemp()


class TestADOwithSQLServer(CommonDBTests):
    def setUp(self):
        self.conn = config.dbSqlServerconnect(
            *config.connStrSQLServer[0], **config.connStrSQLServer[1]
        )
        self.conn.timeout = 30  # turn timeout back up
        self.engine = "MSSQL"
        self.db = config.dbSqlServerconnect
        self.remote = config.connStrSQLServer[2]

    def tearDown(self):
        try:
            self.conn.rollback()
        except:
            pass
        try:
            self.conn.close()
        except:
            pass
        self.conn = None

    def getConnection(self):
        return self.conn

    def getAnotherConnection(self, addkeys=None):
        keys = dict(config.connStrSQLServer[1])
        if addkeys:
            keys.update(addkeys)
        return config.dbSqlServerconnect(*config.connStrSQLServer[0], **keys)

    def testVariableReturningStoredProcedure(self):
        crsr = self.conn.cursor()
        spdef = """
            CREATE PROCEDURE sp_DeleteMeOnlyForTesting
                @theInput varchar(50),
                @theOtherInput varchar(50),
                @theOutput varchar(100) OUTPUT
            AS
                SET @theOutput=@theInput+@theOtherInput
                    """
        try:
            crsr.execute("DROP PROCEDURE sp_DeleteMeOnlyForTesting")
            self.conn.commit()
        except:  # Make sure it is empty
            pass
        crsr.execute(spdef)

        retvalues = crsr.callproc(
            "sp_DeleteMeOnlyForTesting", ("Dodsworth", "Anne", "              ")
        )
        assert retvalues[0] == "Dodsworth", '%s is not "Dodsworth"' % repr(retvalues[0])
        assert retvalues[1] == "Anne", '%s is not "Anne"' % repr(retvalues[1])
        assert retvalues[2] == "DodsworthAnne", '%s is not "DodsworthAnne"' % repr(
            retvalues[2]
        )
        self.conn.rollback()

    def testMultipleSetReturn(self):
        crsr = self.getCursor()
        self.helpCreateAndPopulateTableTemp(crsr)

        spdef = """
            CREATE PROCEDURE sp_DeleteMe_OnlyForTesting
            AS
                SELECT fldData FROM xx_%s ORDER BY fldData ASC
                SELECT fldData From xx_%s where fldData = -9999
                SELECT fldData FROM xx_%s ORDER BY fldData DESC
                    """ % (
            config.tmp,
            config.tmp,
            config.tmp,
        )
        try:
            crsr.execute("DROP PROCEDURE sp_DeleteMe_OnlyForTesting")
            self.conn.commit()
        except:  # Make sure it is empty
            pass
        crsr.execute(spdef)

        retvalues = crsr.callproc("sp_DeleteMe_OnlyForTesting")
        row = crsr.fetchone()
        self.assertEqual(row[0], 0)
        assert crsr.nextset() == True, "Operation should succeed"
        assert not crsr.fetchall(), "Should be an empty second set"
        assert crsr.nextset() == True, "third set should be present"
        rowdesc = crsr.fetchall()
        self.assertEqual(rowdesc[0][0], 8)
        assert crsr.nextset() == None, "No more return sets, should return None"

        self.helpRollbackTblTemp()

    def testDatetimeProcedureParameter(self):
        crsr = self.conn.cursor()
        spdef = """
            CREATE PROCEDURE sp_DeleteMeOnlyForTesting
                @theInput DATETIME,
                @theOtherInput varchar(50),
                @theOutput varchar(100) OUTPUT
            AS
                SET @theOutput = CONVERT(CHARACTER(20), @theInput, 0) + @theOtherInput
                    """
        try:
            crsr.execute("DROP PROCEDURE sp_DeleteMeOnlyForTesting")
            self.conn.commit()
        except:  # Make sure it is empty
            pass
        crsr.execute(spdef)

        result = crsr.callproc(
            "sp_DeleteMeOnlyForTesting",
            [adodbapi.Timestamp(2014, 12, 25, 0, 1, 0), "Beep", " " * 30],
        )

        assert result[2] == "Dec 25 2014 12:01AM Beep", 'value was="%s"' % result[2]
        self.conn.rollback()

    def testIncorrectStoredProcedureParameter(self):
        crsr = self.conn.cursor()
        spdef = """
            CREATE PROCEDURE sp_DeleteMeOnlyForTesting
                @theInput DATETIME,
                @theOtherInput varchar(50),
                @theOutput varchar(100) OUTPUT
            AS
                SET @theOutput = CONVERT(CHARACTER(20), @theInput) + @theOtherInput
                    """
        try:
            crsr.execute("DROP PROCEDURE sp_DeleteMeOnlyForTesting")
            self.conn.commit()
        except:  # Make sure it is empty
            pass
        crsr.execute(spdef)

        # calling the sproc with a string for the first parameter where a DateTime is expected
        result = tryconnection.try_operation_with_expected_exception(
            (api.DataError, api.DatabaseError),
            crsr.callproc,
            ["sp_DeleteMeOnlyForTesting"],
            {"parameters": ["this is wrong", "Anne", "not Alice"]},
        )
        if result[0]:  # the expected exception was raised
            assert "@theInput" in str(result[1]) or "DatabaseError" in str(
                result
            ), "Identifies the wrong erroneous parameter"
        else:
            assert result[0], result[1]  # incorrect or no exception
        self.conn.rollback()


class TestADOwithAccessDB(CommonDBTests):
    def setUp(self):
        self.conn = config.dbAccessconnect(
            *config.connStrAccess[0], **config.connStrAccess[1]
        )
        self.conn.timeout = 30  # turn timeout back up
        self.engine = "ACCESS"
        self.db = config.dbAccessconnect
        self.remote = config.connStrAccess[2]

    def tearDown(self):
        try:
            self.conn.rollback()
        except:
            pass
        try:
            self.conn.close()
        except:
            pass
        self.conn = None

    def getConnection(self):
        return self.conn

    def getAnotherConnection(self, addkeys=None):
        raise NotImplementedError("Jet cannot use a second connection to the database")

    def testOkConnect(self):
        c = self.db(*config.connStrAccess[0], **config.connStrAccess[1])
        assert c != None
        c.close()


class TestADOwithMySql(CommonDBTests):
    def setUp(self):
        self.conn = config.dbMySqlconnect(
            *config.connStrMySql[0], **config.connStrMySql[1]
        )
        self.conn.timeout = 30  # turn timeout back up
        self.engine = "MySQL"
        self.db = config.dbMySqlconnect
        self.remote = config.connStrMySql[2]

    def tearDown(self):
        try:
            self.conn.rollback()
        except:
            pass
        try:
            self.conn.close()
        except:
            pass
        self.conn = None

    def getConnection(self):
        return self.conn

    def getAnotherConnection(self, addkeys=None):
        keys = dict(config.connStrMySql[1])
        if addkeys:
            keys.update(addkeys)
        return config.dbMySqlconnect(*config.connStrMySql[0], **keys)

    def testOkConnect(self):
        c = self.db(*config.connStrMySql[0], **config.connStrMySql[1])
        assert c != None

    # def testStoredProcedure(self):
    #     crsr=self.conn.cursor()
    #     try:
    #         crsr.execute("DROP PROCEDURE DeleteMeOnlyForTesting")
    #         self.conn.commit()
    #     except: #Make sure it is empty
    #         pass
    #     spdef= """
    #             DELIMITER $$
    #             CREATE PROCEDURE DeleteMeOnlyForTesting (onein CHAR(10), twoin CHAR(10), OUT theout CHAR(20))
    #             DETERMINISTIC
    #              BEGIN
    #                 SET theout = onein //|| twoin;
    #                 /* (SELECT 'a small string' as result; */
    #                 END $$
    #             """
    #
    #     crsr.execute(spdef)
    #
    #     retvalues=crsr.callproc('DeleteMeOnlyForTesting',('Dodsworth','Anne','              '))
    #     print 'return value (mysql)=',repr(crsr.returnValue) ###
    #     assert retvalues[0]=='Dodsworth', '%s is not "Dodsworth"'%repr(retvalues[0])
    #     assert retvalues[1]=='Anne','%s is not "Anne"'%repr(retvalues[1])
    #     assert retvalues[2]=='DodsworthAnne','%s is not "DodsworthAnne"'%repr(retvalues[2])
    #
    #     try:
    #         crsr.execute("DROP PROCEDURE, DeleteMeOnlyForTesting")
    #         self.conn.commit()
    #     except: #Make sure it is empty
    #         pass


class TestADOwithPostgres(CommonDBTests):
    def setUp(self):
        self.conn = config.dbPostgresConnect(
            *config.connStrPostgres[0], **config.connStrPostgres[1]
        )
        self.conn.timeout = 30  # turn timeout back up
        self.engine = "PostgreSQL"
        self.db = config.dbPostgresConnect
        self.remote = config.connStrPostgres[2]

    def tearDown(self):
        try:
            self.conn.rollback()
        except:
            pass
        try:
            self.conn.close()
        except:
            pass
        self.conn = None

    def getConnection(self):
        return self.conn

    def getAnotherConnection(self, addkeys=None):
        keys = dict(config.connStrPostgres[1])
        if addkeys:
            keys.update(addkeys)
        return config.dbPostgresConnect(*config.connStrPostgres[0], **keys)

    def testOkConnect(self):
        c = self.db(*config.connStrPostgres[0], **config.connStrPostgres[1])
        assert c != None

    # def testStoredProcedure(self):
    #     crsr=self.conn.cursor()
    #     spdef= """
    #         CREATE OR REPLACE FUNCTION DeleteMeOnlyForTesting (text, text)
    #         RETURNS text AS $funk$
    #         BEGIN
    #           RETURN $1 || $2;
    #         END;
    #         $funk$
    #         LANGUAGE SQL;
    #         """
    #
    #     crsr.execute(spdef)
    #     retvalues = crsr.callproc('DeleteMeOnlyForTesting',('Dodsworth','Anne','              '))
    #     ### print 'return value (pg)=',repr(crsr.returnValue) ###
    #     assert retvalues[0]=='Dodsworth', '%s is not "Dodsworth"'%repr(retvalues[0])
    #     assert retvalues[1]=='Anne','%s is not "Anne"'%repr(retvalues[1])
    #     assert retvalues[2]=='Dodsworth Anne','%s is not "Dodsworth Anne"'%repr(retvalues[2])
    #     self.conn.rollback()
    #     try:
    #         crsr.execute("DROP PROCEDURE, DeleteMeOnlyForTesting")
    #         self.conn.commit()
    #     except: #Make sure it is empty
    #         pass


class TimeConverterInterfaceTest(unittest.TestCase):
    def testIDate(self):
        assert self.tc.Date(1990, 2, 2)

    def testITime(self):
        assert self.tc.Time(13, 2, 2)

    def testITimestamp(self):
        assert self.tc.Timestamp(1990, 2, 2, 13, 2, 1)

    def testIDateObjectFromCOMDate(self):
        assert self.tc.DateObjectFromCOMDate(37435.7604282)

    def testICOMDate(self):
        assert hasattr(self.tc, "COMDate")

    def testExactDate(self):
        d = self.tc.Date(1994, 11, 15)
        comDate = self.tc.COMDate(d)
        correct = 34653.0
        assert comDate == correct, comDate

    def testExactTimestamp(self):
        d = self.tc.Timestamp(1994, 11, 15, 12, 0, 0)
        comDate = self.tc.COMDate(d)
        correct = 34653.5
        self.assertEqual(comDate, correct)

        d = self.tc.Timestamp(2003, 5, 6, 14, 15, 17)
        comDate = self.tc.COMDate(d)
        correct = 37747.593946759262
        self.assertEqual(comDate, correct)

    def testIsoFormat(self):
        d = self.tc.Timestamp(1994, 11, 15, 12, 3, 10)
        iso = self.tc.DateObjectToIsoFormatString(d)
        self.assertEqual(str(iso[:19]), "1994-11-15 12:03:10")

        dt = self.tc.Date(2003, 5, 2)
        iso = self.tc.DateObjectToIsoFormatString(dt)
        self.assertEqual(str(iso[:10]), "2003-05-02")


if config.doMxDateTimeTest:
    import mx.DateTime


class TestMXDateTimeConverter(TimeConverterInterfaceTest):
    def setUp(self):
        self.tc = api.mxDateTimeConverter()

    def testCOMDate(self):
        t = mx.DateTime.DateTime(2002, 6, 28, 18, 15, 2)
        cmd = self.tc.COMDate(t)
        assert cmd == t.COMDate()

    def testDateObjectFromCOMDate(self):
        cmd = self.tc.DateObjectFromCOMDate(37435.7604282)
        t = mx.DateTime.DateTime(2002, 6, 28, 18, 15, 0)
        t2 = mx.DateTime.DateTime(2002, 6, 28, 18, 15, 2)
        assert t2 > cmd > t

    def testDate(self):
        assert mx.DateTime.Date(1980, 11, 4) == self.tc.Date(1980, 11, 4)

    def testTime(self):
        assert mx.DateTime.Time(13, 11, 4) == self.tc.Time(13, 11, 4)

    def testTimestamp(self):
        t = mx.DateTime.DateTime(2002, 6, 28, 18, 15, 1)
        obj = self.tc.Timestamp(2002, 6, 28, 18, 15, 1)
        assert t == obj


import time


class TestPythonTimeConverter(TimeConverterInterfaceTest):
    def setUp(self):
        self.tc = api.pythonTimeConverter()

    def testCOMDate(self):
        mk = time.mktime((2002, 6, 28, 18, 15, 1, 4, 31 + 28 + 31 + 30 + 31 + 28, -1))
        t = time.localtime(mk)
        # Fri, 28 Jun 2002 18:15:01 +0000
        cmd = self.tc.COMDate(t)
        assert abs(cmd - 37435.7604282) < 1.0 / 24, "%f more than an hour wrong" % cmd

    def testDateObjectFromCOMDate(self):
        cmd = self.tc.DateObjectFromCOMDate(37435.7604282)
        t1 = time.gmtime(
            time.mktime((2002, 6, 28, 0, 14, 1, 4, 31 + 28 + 31 + 30 + 31 + 28, -1))
        )
        # there are errors in the implementation of gmtime which we ignore
        t2 = time.gmtime(
            time.mktime((2002, 6, 29, 12, 14, 2, 4, 31 + 28 + 31 + 30 + 31 + 28, -1))
        )
        assert t1 < cmd < t2, '"%s" should be about 2002-6-28 12:15:01' % repr(cmd)

    def testDate(self):
        t1 = time.mktime((2002, 6, 28, 18, 15, 1, 4, 31 + 28 + 31 + 30 + 31 + 30, 0))
        t2 = time.mktime((2002, 6, 30, 18, 15, 1, 4, 31 + 28 + 31 + 30 + 31 + 28, 0))
        obj = self.tc.Date(2002, 6, 29)
        assert t1 < time.mktime(obj) < t2, obj

    def testTime(self):
        self.assertEqual(
            self.tc.Time(18, 15, 2), time.gmtime(18 * 60 * 60 + 15 * 60 + 2)
        )

    def testTimestamp(self):
        t1 = time.localtime(
            time.mktime((2002, 6, 28, 18, 14, 1, 4, 31 + 28 + 31 + 30 + 31 + 28, -1))
        )
        t2 = time.localtime(
            time.mktime((2002, 6, 28, 18, 16, 1, 4, 31 + 28 + 31 + 30 + 31 + 28, -1))
        )
        obj = self.tc.Timestamp(2002, 6, 28, 18, 15, 2)
        assert t1 < obj < t2, obj


class TestPythonDateTimeConverter(TimeConverterInterfaceTest):
    def setUp(self):
        self.tc = api.pythonDateTimeConverter()

    def testCOMDate(self):
        t = datetime.datetime(2002, 6, 28, 18, 15, 1)
        # Fri, 28 Jun 2002 18:15:01 +0000
        cmd = self.tc.COMDate(t)
        assert abs(cmd - 37435.7604282) < 1.0 / 24, "more than an hour wrong"

    def testDateObjectFromCOMDate(self):
        cmd = self.tc.DateObjectFromCOMDate(37435.7604282)
        t1 = datetime.datetime(2002, 6, 28, 18, 14, 1)
        t2 = datetime.datetime(2002, 6, 28, 18, 16, 1)
        assert t1 < cmd < t2, cmd

        tx = datetime.datetime(
            2002, 6, 28, 18, 14, 1, 900000
        )  # testing that microseconds don't become milliseconds
        c1 = self.tc.DateObjectFromCOMDate(self.tc.COMDate(tx))
        assert t1 < c1 < t2, c1

    def testDate(self):
        t1 = datetime.date(2002, 6, 28)
        t2 = datetime.date(2002, 6, 30)
        obj = self.tc.Date(2002, 6, 29)
        assert t1 < obj < t2, obj

    def testTime(self):
        self.assertEqual(self.tc.Time(18, 15, 2).isoformat()[:8], "18:15:02")

    def testTimestamp(self):
        t1 = datetime.datetime(2002, 6, 28, 18, 14, 1)
        t2 = datetime.datetime(2002, 6, 28, 18, 16, 1)
        obj = self.tc.Timestamp(2002, 6, 28, 18, 15, 2)
        assert t1 < obj < t2, obj


suites = []
suites.append(unittest.makeSuite(TestPythonDateTimeConverter, "test"))
if config.doMxDateTimeTest:
    suites.append(unittest.makeSuite(TestMXDateTimeConverter, "test"))
if config.doTimeTest:
    suites.append(unittest.makeSuite(TestPythonTimeConverter, "test"))

if config.doAccessTest:
    suites.append(unittest.makeSuite(TestADOwithAccessDB, "test"))
if config.doSqlServerTest:
    suites.append(unittest.makeSuite(TestADOwithSQLServer, "test"))
if config.doMySqlTest:
    suites.append(unittest.makeSuite(TestADOwithMySql, "test"))
if config.doPostgresTest:
    suites.append(unittest.makeSuite(TestADOwithPostgres, "test"))


class cleanup_manager(object):
    def __enter__(self):
        pass

    def __exit__(self, exc_type, exc_val, exc_tb):
        config.cleanup(config.testfolder, config.mdb_name)


suite = unittest.TestSuite(suites)
if __name__ == "__main__":
    mysuite = copy.deepcopy(suite)
    with cleanup_manager():
        defaultDateConverter = adodbapi.dateconverter
        print(__doc__)
        print("Default Date Converter is %s" % (defaultDateConverter,))
        dateconverter = defaultDateConverter
        tag = "datetime"
        unittest.TextTestRunner().run(mysuite)

        if config.iterateOverTimeTests:
            for test, dateconverter, tag in (
                (config.doTimeTest, api.pythonTimeConverter, "pythontime"),
                (config.doMxDateTimeTest, api.mxDateTimeConverter, "mx"),
            ):
                if test:
                    mysuite = copy.deepcopy(
                        suite
                    )  # work around a side effect of unittest.TextTestRunner
                    adodbapi.adodbapi.dateconverter = dateconverter()
                    print("Changed dateconverter to ")
                    print(adodbapi.adodbapi.dateconverter)
                    unittest.TextTestRunner().run(mysuite)