TestCase for pyodbc driver BLOB support

1 files changed, 49 insertions, 0 deletions

diff --git a/test/unittest_db.py b/test/unittest_db.py
index af288ff..183cdf4 100644
--- a/test/unittest_db.py
+++ b/test/unittest_db.py
@@ -231,5 +231,54 @@ class DBAPIAdaptersTC(TestCase):
         self.assertEquals(slhelper.func_sqlname('MYFUNC'), 'SQLITE_MYFUNC')
 
 
+class pyodbcTC(TestCase):
+    def setUp(self):
+        import pyodbc
+        try:
+            self.cnx = get_connection(driver='sqlserver2005', database='alf', 
+                                  host='localhost', extra_args='Trusted_Connection')
+        except pyodbc.Error, exc:
+            self.skip(str(exc))
+        cursor = self.cnx.cursor()
+        try:
+            cursor.execute('create table TestLargeString (id int, data varchar(max))')
+            cursor.execute('create table TestBlob (id int, data varbinary(max))')
+        except Exception, exc:
+            print exc
+        cursor.close()
+
+    def tearDown(self):
+        cursor = self.cnx.cursor()
+        cursor.execute('drop table TestBlob')
+        cursor.execute('drop table TestLargeString')
+        cursor.close()
+        self.cnx.close()
+
+    def test_blob(self):
+        cursor = self.cnx.cursor()
+        data_length = xrange(400*1024-10, 400*1024+10)
+        for length in data_length:
+            data = buffer('\x00'*length)
+            print "inserting string of length", len(data)
+            cursor.execute('insert into TestBlob(id, data) VALUES(%(id)s, %(data)s)', 
+                           {'id': length, 'data': data})
+            self.cnx.commit()
+        cursor.execute('select count(*) from TestBlob')
+        print '%d rows in table' % (cursor.fetchone()[0])
+        cursor.close()
+
+    def test_large_string(self):
+        cursor = self.cnx.cursor()
+        data_length = xrange(400*1024-10, 400*1024+10)
+        for length in data_length:
+            data = '1'*length
+            print "inserting string of length", len(data)
+            cursor.execute('insert into TestLargeString(id, data) VALUES(%(id)s, %(data)s)', 
+                           {'id': length, 'data': data})
+            self.cnx.commit()
+        cursor.execute('select count(*) from TestLargeString')
+        print '%d rows in table' % (cursor.fetchone()[0])
+        cursor.close()
+
 if __name__ == '__main__':
     unittest_main()