install pywin32-220

Lib/site-packages/win32/test/test_odbc.py

@@ -0,0 +1,192 @@
+# odbc test suite kindly contributed by Frank Millman.
+import sys
+import os
+import unittest
+import odbc
+import tempfile
+
+from pywin32_testutil import str2bytes, str2memory, TestSkipped
+
+# We use the DAO ODBC driver
+from win32com.client.gencache import EnsureDispatch
+from win32com.client import constants
+import pythoncom
+
+class TestStuff(unittest.TestCase):
+    def setUp(self):
+        self.tablename = "pywin32test_users"
+        self.db_filename = None
+        self.conn = self.cur = None
+        try:
+            # Test any database if a connection string is supplied...
+            conn_str = os.environ['TEST_ODBC_CONNECTION_STRING']
+        except KeyError:
+            # Create a local MSAccess DB for testing.
+            self.db_filename = tempfile.NamedTemporaryFile().name + '.mdb'
+    
+            # Create a brand-new database - what is the story with these?
+            for suffix in (".36", ".35", ".30"):
+                try:
+                    dbe = EnsureDispatch("DAO.DBEngine" + suffix)
+                    break
+                except pythoncom.com_error:
+                    pass
+            else:
+                raise TestSkipped("Can't find a DB engine")
+    
+            workspace = dbe.Workspaces(0)
+    
+            newdb = workspace.CreateDatabase(self.db_filename, 
+                                             constants.dbLangGeneral,
+                                             constants.dbEncrypt)
+    
+            newdb.Close()
+    
+            conn_str = "Driver={Microsoft Access Driver (*.mdb)};dbq=%s;Uid=;Pwd=;" \
+                       % (self.db_filename,)
+        ## print 'Connection string:', conn_str
+        self.conn = odbc.odbc(conn_str)
+        # And we expect a 'users' table for these tests.
+        self.cur = self.conn.cursor()
+        ## self.cur.setoutputsize(1000)
+        try:
+            self.cur.execute("""drop table %s""" %self.tablename)
+        except (odbc.error, odbc.progError):
+            pass
+
+        ## This needs to be adjusted for sql server syntax for unicode fields
+        ##  - memo -> TEXT
+        ##  - varchar -> nvarchar
+        self.assertEqual(self.cur.execute(
+            """create table %s (
+                    userid varchar(25),
+                    username varchar(25),
+                    bitfield bit,
+                    intfield integer,
+                    floatfield float,
+                    datefield datetime,
+                    rawfield varbinary(100),
+                    longtextfield memo,
+                    longbinaryfield image
+            )""" %self.tablename),-1)
+
+    def tearDown(self):
+        if self.cur is not None:
+            try:
+                self.cur.execute("""drop table %s""" %self.tablename)
+            except (odbc.error, odbc.progError) as why:
+                print("Failed to delete test table %s" %self.tablename, why)
+
+            self.cur.close()
+            self.cur = None
+        if self.conn is not None:
+            self.conn.close()
+            self.conn = None
+        if self.db_filename is not None:
+            try:
+                os.unlink(self.db_filename)
+            except OSError:
+                pass
+
+    def test_insert_select(self, userid='Frank', username='Frank Millman'):
+        self.assertEqual(self.cur.execute("insert into %s (userid, username) \
+            values (?,?)" %self.tablename, [userid, username]),1)
+        self.assertEqual(self.cur.execute("select * from %s \
+            where userid = ?" %self.tablename, [userid.lower()]),0)
+        self.assertEqual(self.cur.execute("select * from %s \
+            where username = ?" %self.tablename, [username.lower()]),0)
+
+    def test_insert_select_unicode(self, userid='Frank', username="Frank Millman"):
+        self.assertEqual(self.cur.execute("insert into %s (userid, username)\
+            values (?,?)" %self.tablename, [userid, username]),1)
+        self.assertEqual(self.cur.execute("select * from %s \
+            where userid = ?" %self.tablename, [userid.lower()]),0)
+        self.assertEqual(self.cur.execute("select * from %s \
+            where username = ?" %self.tablename, [username.lower()]),0)
+
+    def test_insert_select_unicode_ext(self):
+        userid = "t-\xe0\xf2"
+        username = "test-\xe0\xf2 name"
+        self.test_insert_select_unicode(userid, username)
+
+    def _test_val(self, fieldName, value):
+        for x in range(100):
+            self.cur.execute("delete from %s where userid='Frank'" %self.tablename)
+            self.assertEqual(self.cur.execute(
+                "insert into %s (userid, %s) values (?,?)" % (self.tablename, fieldName),
+                ["Frank", value]), 1)
+            self.cur.execute("select %s from %s where userid = ?" % (fieldName, self.tablename),
+                             ["Frank"])
+            rows = self.cur.fetchmany()
+            self.failUnlessEqual(1, len(rows))
+            row = rows[0]
+            self.failUnlessEqual(row[0], value)
+
+    def testBit(self):
+        self._test_val('bitfield', 1)
+        self._test_val('bitfield', 0)
+
+    def testInt(self):
+        self._test_val('intfield', 1)
+        self._test_val('intfield', 0)
+        try:
+            big = sys.maxsize
+        except AttributeError:
+            big = sys.maxint
+        self._test_val('intfield', big)
+        
+    def testFloat(self):
+        self._test_val('floatfield', 1.01)
+        self._test_val('floatfield', 0)
+
+    def testVarchar(self, ):
+        self._test_val('username', 'foo')
+
+    def testLongVarchar(self):
+        """ Test a long text field in excess of internal cursor data size (65536)"""
+        self._test_val('longtextfield', 'abc' * 70000)
+
+    def testLongBinary(self):
+        """ Test a long raw field in excess of internal cursor data size (65536)"""
+        self._test_val('longbinaryfield', str2memory('\0\1\2' * 70000))
+
+    def testRaw(self):
+        ## Test binary data
+        self._test_val('rawfield', str2memory('\1\2\3\4\0\5\6\7\8'))
+
+    def test_widechar(self):
+        """Test a unicode character that would be mangled if bound as plain character.
+            For example, previously the below was returned as ascii 'a'
+        """
+        self._test_val('username', '\u0101')
+
+    def testDates(self):
+        import datetime
+        for v in (
+            (1900, 12, 25, 23, 39, 59),
+            ):
+            d = datetime.datetime(*v)
+            self._test_val('datefield', d)
+
+    def test_set_nonzero_length(self):
+        self.assertEqual(self.cur.execute("insert into %s (userid,username) "
+            "values (?,?)" %self.tablename, ['Frank', 'Frank Millman']),1)
+        self.assertEqual(self.cur.execute("update %s set username = ?" %self.tablename,
+            ['Frank']),1)
+        self.assertEqual(self.cur.execute("select * from %s" %self.tablename), 0)
+        self.assertEqual(len(self.cur.fetchone()[1]),5)
+
+    def test_set_zero_length(self):
+        self.assertEqual(self.cur.execute("insert into %s (userid,username) "
+            "values (?,?)" %self.tablename, [str2bytes('Frank'), '']),1)
+        self.assertEqual(self.cur.execute("select * from %s" %self.tablename), 0)
+        self.assertEqual(len(self.cur.fetchone()[1]),0)
+
+    def test_set_zero_length_unicode(self):
+        self.assertEqual(self.cur.execute("insert into %s (userid,username) "
+            "values (?,?)" %self.tablename, ['Frank', '']),1)
+        self.assertEqual(self.cur.execute("select * from %s" %self.tablename), 0)
+        self.assertEqual(len(self.cur.fetchone()[1]),0)
+
+if __name__ == '__main__':
+    unittest.main()