# 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), 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=u'Frank', username=u"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 = u"t-\xe0\xf2" username = u"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', u'\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, [u'Frank', u'']),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()