My favorites | Sign in
Project Home Downloads Wiki Issues Source
Repository:
Checkout   Browse   Changes   Clones  
Changes to /tests3/sqlitetests.py
c278e74e4434 vs. b2dfac9a0b88 Compare: vs.  Format:
Revision b2dfac9a0b88
Go to: 
Sign in to write a code review
/tests3/sqlitetests.py   c278e74e4434 /tests3/sqlitetests.py   b2dfac9a0b88
1 #!/usr/bin/python 1 #!/usr/bin/python
2 # -*- coding: latin-1 -*- 2 # -*- coding: latin-1 -*-
3 3
4 usage = """\ 4 usage = """\
5 usage: %prog [options] connection_string 5 usage: %prog [options] connection_string
6 6
7 Unit tests for SQLite using the ODBC driver from http://www.ch-werner.de/sqliteodbc 7 Unit tests for SQLite using the ODBC driver from http://www.ch-werner.de/sqliteodbc
8 8
9 To use, pass a connection string as the parameter. The tests will create and 9 To use, pass a connection string as the parameter. The tests will create and
10 drop tables t1 and t2 as necessary. On Windows, use the 32-bit driver with 10 drop tables t1 and t2 as necessary. On Windows, use the 32-bit driver with
11 32-bit Python and the 64-bit driver with 64-bit Python (regardless of your 11 32-bit Python and the 64-bit driver with 64-bit Python (regardless of your
12 operating system bitness). 12 operating system bitness).
13 13
14 These run using the version from the 'build' directory, not the version 14 These run using the version from the 'build' directory, not the version
15 installed into the Python directories. You must run python setup.py build 15 installed into the Python directories. You must run python setup.py build
16 before running the tests. 16 before running the tests.
17 17
18 You can also put the connection string into a setup.cfg file in the root of the project 18 You can also put the connection string into a setup.cfg file in the root of the project
19 (the same one setup.py would use) like so: 19 (the same one setup.py would use) like so:
20 20
21 [sqlitetests] 21 [sqlitetests]
22 connection-string=Driver=SQLite3 ODBC Driver;Database=sqlite.db 22 connection-string=Driver=SQLite3 ODBC Driver;Database=sqlite.db
23 """ 23 """
24 24
25 import sys, os, re 25 import sys, os, re
26 import unittest 26 import unittest
27 from decimal import Decimal 27 from decimal import Decimal
28 from datetime import datetime, date, time 28 from datetime import datetime, date, time
29 from os.path import join, getsize, dirname, abspath 29 from os.path import join, getsize, dirname, abspath
30 from testutils import * 30 from testutils import *
31 31
32 _TESTSTR = '0123456789-abcdefghijklmnopqrstuvwxyz-' 32 _TESTSTR = '0123456789-abcdefghijklmnopqrstuvwxyz-'
33 33
34 def _generate_test_string(length): 34 def _generate_test_string(length):
35 """ 35 """
36 Returns a string of `length` characters, constructed by repeating _TESTSTR as necessary. 36 Returns a string of `length` characters, constructed by repeating _TESTSTR as necessary.
37 37
38 To enhance performance, there are 3 ways data is read, based on the length of the value, so most data types are 38 To enhance performance, there are 3 ways data is read, based on the length of the value, so most data types are
39 tested with 3 lengths. This function helps us generate the test data. 39 tested with 3 lengths. This function helps us generate the test data.
40 40
41 We use a recognizable data set instead of a single character to make it less likely that "overlap" errors will 41 We use a recognizable data set instead of a single character to make it less likely that "overlap" errors will
42 be hidden and to help us manually identify where a break occurs. 42 be hidden and to help us manually identify where a break occurs.
43 """ 43 """
44 if length <= len(_TESTSTR): 44 if length <= len(_TESTSTR):
45 return _TESTSTR[:length] 45 return _TESTSTR[:length]
46 46
47 c = (length + len(_TESTSTR)-1) / len(_TESTSTR) 47 c = (length + len(_TESTSTR)-1) // len(_TESTSTR)
48 v = _TESTSTR * c 48 v = _TESTSTR * c
49 return v[:length] 49 return v[:length]
50 50
51 class SqliteTestCase(unittest.TestCase): 51 class SqliteTestCase(unittest.TestCase):
52 52
53 SMALL_FENCEPOST_SIZES = [ 0, 1, 255, 256, 510, 511, 512, 1023, 1024, 2047, 2048, 4000 ] 53 SMALL_FENCEPOST_SIZES = [ 0, 1, 255, 256, 510, 511, 512, 1023, 1024, 2047, 2048, 4000 ]
54 LARGE_FENCEPOST_SIZES = [ 4095, 4096, 4097, 10 * 1024, 20 * 1024 ] 54 LARGE_FENCEPOST_SIZES = [ 4095, 4096, 4097, 10 * 1024, 20 * 1024 ]
55 55
56 ANSI_FENCEPOSTS = [ _generate_test_string(size) for size in SMALL_FENCEPOST_SIZES ] 56 STR_FENCEPOSTS = [ _generate_test_string(size) for size in SMALL_FENCEPOST_SIZES ]
57 UNICODE_FENCEPOSTS = [ unicode(s) for s in ANSI_FENCEPOSTS ] 57 BYTE_FENCEPOSTS = [ bytes(s, 'ascii') for s in STR_FENCEPOSTS ]
58 IMAGE_FENCEPOSTS = ANSI_FENCEPOSTS + [ _generate_test_string(size) for size in LARGE_FENCEPOST_SIZES ] 58 IMAGE_FENCEPOSTS = BYTE_FENCEPOSTS + [ bytes(_generate_test_string(size), 'ascii') for size in LARGE_FENCEPOST_SIZES ]
59 59
60 def __init__(self, method_name, connection_string): 60 def __init__(self, method_name, connection_string):
61 unittest.TestCase.__init__(self, method_name) 61 unittest.TestCase.__init__(self, method_name)
62 self.connection_string = connection_string 62 self.connection_string = connection_string
63 63
64 def get_sqlite_version(self): 64 def get_sqlite_version(self):
65 """ 65 """
66 Returns the major version: 8-->2000, 9-->2005, 10-->2008 66 Returns the major version: 8-->2000, 9-->2005, 10-->2008
67 """ 67 """
68 self.cursor.execute("exec master..xp_msver 'ProductVersion'") 68 self.cursor.execute("exec master..xp_msver 'ProductVersion'")
69 row = self.cursor.fetchone() 69 row = self.cursor.fetchone()
70 return int(row.Character_Value.split('.', 1)[0]) 70 return int(row.Character_Value.split('.', 1)[0])
71 71
72 def setUp(self): 72 def setUp(self):
73 self.cnxn = pyodbc.connect(self.connection_string) 73 self.cnxn = pyodbc.connect(self.connection_string)
74 self.cursor = self.cnxn.cursor() 74 self.cursor = self.cnxn.cursor()
75 75
76 for i in range(3): 76 for i in range(3):
77 try: 77 try:
78 self.cursor.execute("drop table t%d" % i) 78 self.cursor.execute("drop table t%d" % i)
79 self.cnxn.commit() 79 self.cnxn.commit()
80 except: 80 except:
81 pass 81 pass
82 82
83 self.cnxn.rollback() 83 self.cnxn.rollback()
84 84
85 def tearDown(self): 85 def tearDown(self):
86 try: 86 try:
87 self.cursor.close() 87 self.cursor.close()
88 self.cnxn.close() 88 self.cnxn.close()
89 except: 89 except:
90 # If we've already closed the cursor or connection, exceptions are thrown. 90 # If we've already closed the cursor or connection, exceptions are thrown.
91 pass 91 pass
92 92
93 def test_multiple_bindings(self): 93 def test_multiple_bindings(self):
94 "More than one bind and select on a cursor" 94 "More than one bind and select on a cursor"
95 self.cursor.execute("create table t1(n int)") 95 self.cursor.execute("create table t1(n int)")
96 self.cursor.execute("insert into t1 values (?)", 1) 96 self.cursor.execute("insert into t1 values (?)", 1)
97 self.cursor.execute("insert into t1 values (?)", 2) 97 self.cursor.execute("insert into t1 values (?)", 2)
98 self.cursor.execute("insert into t1 values (?)", 3) 98 self.cursor.execute("insert into t1 values (?)", 3)
99 for i in range(3): 99 for i in range(3):
100 self.cursor.execute("select n from t1 where n < ?", 10) 100 self.cursor.execute("select n from t1 where n < ?", 10)
101 self.cursor.execute("select n from t1 where n < 3") 101 self.cursor.execute("select n from t1 where n < 3")
102 102
103 103
104 def test_different_bindings(self): 104 def test_different_bindings(self):
105 self.cursor.execute("create table t1(n int)") 105 self.cursor.execute("create table t1(n int)")
106 self.cursor.execute("create table t2(d datetime)") 106 self.cursor.execute("create table t2(d datetime)")
107 self.cursor.execute("insert into t1 values (?)", 1) 107 self.cursor.execute("insert into t1 values (?)", 1)
108 self.cursor.execute("insert into t2 values (?)", datetime.now()) 108 self.cursor.execute("insert into t2 values (?)", datetime.now())
109 109
110 def test_datasources(self): 110 def test_datasources(self):
111 p = pyodbc.dataSources() 111 p = pyodbc.dataSources()
112 self.assert_(isinstance(p, dict)) 112 self.assert_(isinstance(p, dict))
113 113
114 def test_getinfo_string(self): 114 def test_getinfo_string(self):
115 value = self.cnxn.getinfo(pyodbc.SQL_CATALOG_NAME_SEPARATOR) 115 value = self.cnxn.getinfo(pyodbc.SQL_CATALOG_NAME_SEPARATOR)
116 self.assert_(isinstance(value, str)) 116 self.assert_(isinstance(value, str))
117 117
118 def test_getinfo_bool(self): 118 def test_getinfo_bool(self):
119 value = self.cnxn.getinfo(pyodbc.SQL_ACCESSIBLE_TABLES) 119 value = self.cnxn.getinfo(pyodbc.SQL_ACCESSIBLE_TABLES)
120 self.assert_(isinstance(value, bool)) 120 self.assert_(isinstance(value, bool))
121 121
122 def test_getinfo_int(self): 122 def test_getinfo_int(self):
123 value = self.cnxn.getinfo(pyodbc.SQL_DEFAULT_TXN_ISOLATION) 123 value = self.cnxn.getinfo(pyodbc.SQL_DEFAULT_TXN_ISOLATION)
124 self.assert_(isinstance(value, (int, long))) 124 self.assert_(isinstance(value, int))
125 125
126 def test_getinfo_smallint(self): 126 def test_getinfo_smallint(self):
127 value = self.cnxn.getinfo(pyodbc.SQL_CONCAT_NULL_BEHAVIOR) 127 value = self.cnxn.getinfo(pyodbc.SQL_CONCAT_NULL_BEHAVIOR)
128 self.assert_(isinstance(value, int)) 128 self.assert_(isinstance(value, int))
129 129
130 def test_fixed_unicode(self):
131 value = u"t\xebsting"
132 self.cursor.execute("create table t1(s nchar(7))")
133 self.cursor.execute("insert into t1 values(?)", u"t\xebsting")
134 v = self.cursor.execute("select * from t1").fetchone()[0]
135 self.assertEqual(type(v), unicode)
136 self.assertEqual(len(v), len(value)) # If we alloc'd wrong, the test below might work because of an embedded NULL
137 self.assertEqual(v, value)
138
139
140 def _test_strtype(self, sqltype, value, colsize=None): 130 def _test_strtype(self, sqltype, value, colsize=None):
141 """ 131 """
142 The implementation for string, Unicode, and binary tests. 132 The implementation for string, Unicode, and binary tests.
143 """ 133 """
144 assert colsize is None or (value is None or colsize >= len(value)) 134 assert colsize is None or (value is None or colsize >= len(value))
145 135
146 if colsize: 136 if colsize:
147 sql = "create table t1(s %s(%s))" % (sqltype, colsize) 137 sql = "create table t1(s %s(%s))" % (sqltype, colsize)
148 else: 138 else:
149 sql = "create table t1(s %s)" % sqltype 139 sql = "create table t1(s %s)" % sqltype
150 140
151 self.cursor.execute(sql) 141 self.cursor.execute(sql)
152 self.cursor.execute("insert into t1 values(?)", value) 142 self.cursor.execute("insert into t1 values(?)", value)
153 v = self.cursor.execute("select * from t1").fetchone()[0] 143 v = self.cursor.execute("select * from t1").fetchone()[0]
154 self.assertEqual(type(v), type(value)) 144 self.assertEqual(type(v), type(value))
155 145
156 if value is not None: 146 if value is not None:
157 self.assertEqual(len(v), len(value)) 147 self.assertEqual(len(v), len(value))
158 148
159 self.assertEqual(v, value) 149 self.assertEqual(v, value)
160 150
161 # Reported by Andy Hochhaus in the pyodbc group: In 2.1.7 and earlier, a hardcoded length of 255 was used to 151 # Reported by Andy Hochhaus in the pyodbc group: In 2.1.7 and earlier, a hardcoded length of 255 was used to
162 # determine whether a parameter was bound as a SQL_VARCHAR or SQL_LONGVARCHAR. Apparently SQL Server chokes if 152 # determine whether a parameter was bound as a SQL_VARCHAR or SQL_LONGVARCHAR. Apparently SQL Server chokes if
163 # we bind as a SQL_LONGVARCHAR and the target column size is 8000 or less, which is considers just SQL_VARCHAR. 153 # we bind as a SQL_LONGVARCHAR and the target column size is 8000 or less, which is considers just SQL_VARCHAR.
164 # This means binding a 256 character value would cause problems if compared with a VARCHAR column under 154 # This means binding a 256 character value would cause problems if compared with a VARCHAR column under
165 # 8001. We now use SQLGetTypeInfo to determine the time to switch. 155 # 8001. We now use SQLGetTypeInfo to determine the time to switch.
166 # 156 #
167 # [42000] [Microsoft][SQL Server Native Client 10.0][SQL Server]The data types varchar and text are incompatible in the equal to operator. 157 # [42000] [Microsoft][SQL Server Native Client 10.0][SQL Server]The data types varchar and text are incompatible in the equal to operator.
168 158
169 self.cursor.execute("select * from t1 where s=?", value) 159 self.cursor.execute("select * from t1 where s=?", value)
170 160
171 161
172 def _test_strliketype(self, sqltype, value, colsize=None): 162 def _test_strliketype(self, sqltype, value, colsize=None):
173 """ 163 """
174 The implementation for text, image, ntext, and binary. 164 The implementation for text, image, ntext, and binary.
175 165
176 These types do not support comparison operators. 166 These types do not support comparison operators.
177 """ 167 """
178 assert colsize is None or (value is None or colsize >= len(value)) 168 assert colsize is None or (value is None or colsize >= len(value))
179 169
180 if colsize: 170 if colsize:
181 sql = "create table t1(s %s(%s))" % (sqltype, colsize) 171 sql = "create table t1(s %s(%s))" % (sqltype, colsize)
182 else: 172 else:
183 sql = "create table t1(s %s)" % sqltype 173 sql = "create table t1(s %s)" % sqltype
184 174
185 self.cursor.execute(sql) 175 self.cursor.execute(sql)
186 self.cursor.execute("insert into t1 values(?)", value) 176 self.cursor.execute("insert into t1 values(?)", value)
187 v = self.cursor.execute("select * from t1").fetchone()[0] 177 v = self.cursor.execute("select * from t1").fetchone()[0]
188 self.assertEqual(type(v), type(value)) 178 self.assertEqual(type(v), type(value))
189 179
190 if value is not None: 180 if value is not None:
191 self.assertEqual(len(v), len(value)) 181 self.assertEqual(len(v), len(value))
192 182
193 self.assertEqual(v, value) 183 self.assertEqual(v, value)
194 184
195 # 185 #
196 # text 186 # text
197 # 187 #
198 188
199 def test_text_null(self): 189 def test_text_null(self):
200 self._test_strtype('text', None, 100) 190 self._test_strtype('text', None, 100)
201 191
202 # Generate a test for each fencepost size: test_text_0, etc. 192 # Generate a test for each fencepost size: test_text_0, etc.
203 def _maketest(value): 193 def _maketest(value):
204 def t(self): 194 def t(self):
205 self._test_strtype('text', value, len(value)) 195 self._test_strtype('text', value, len(value))
206 return t 196 return t
207 for value in UNICODE_FENCEPOSTS: 197 for value in STR_FENCEPOSTS:
208 locals()['test_text_%s' % len(value)] = _maketest(value) 198 locals()['test_text_%s' % len(value)] = _maketest(value)
209 199
210 def test_text_upperlatin(self): 200 def test_text_upperlatin(self):
211 self._test_strtype('varchar', u'á') 201 self._test_strtype('varchar', 'á')
212 202
213 # 203 #
214 # blob 204 # blob
215 # 205 #
216 206
217 def test_null_blob(self): 207 def test_null_blob(self):
218 self._test_strtype('blob', None, 100) 208 self._test_strtype('blob', None, 100)
219 209
220 def test_large_null_blob(self): 210 def test_large_null_blob(self):
221 # Bug 1575064 211 # Bug 1575064
222 self._test_strtype('blob', None, 4000) 212 self._test_strtype('blob', None, 4000)
223 213
224 # Generate a test for each fencepost size: test_unicode_0, etc. 214 # Generate a test for each fencepost size: test_unicode_0, etc.
225 def _maketest(value): 215 def _maketest(value):
226 def t(self): 216 def t(self):
227 self._test_strtype('blob', buffer(value), len(value)) 217 self._test_strtype('blob', bytearray(value), len(value))
228 return t 218 return t
229 for value in ANSI_FENCEPOSTS: 219 for value in BYTE_FENCEPOSTS:
230 locals()['test_blob_%s' % len(value)] = _maketest(value) 220 locals()['test_blob_%s' % len(value)] = _maketest(value)
231 221
232 def test_subquery_params(self): 222 def test_subquery_params(self):
233 """Ensure parameter markers work in a subquery""" 223 """Ensure parameter markers work in a subquery"""
234 self.cursor.execute("create table t1(id integer, s varchar(20))") 224 self.cursor.execute("create table t1(id integer, s varchar(20))")
235 self.cursor.execute("insert into t1 values (?,?)", 1, 'test') 225 self.cursor.execute("insert into t1 values (?,?)", 1, 'test')
236 row = self.cursor.execute(""" 226 row = self.cursor.execute("""
237 select x.id 227 select x.id
238 from ( 228 from (
239 select id 229 select id
240 from t1 230 from t1
241 where s = ? 231 where s = ?
242 and id between ? and ? 232 and id between ? and ?
243 ) x 233 ) x
244 """, 'test', 1, 10).fetchone() 234 """, 'test', 1, 10).fetchone()
245 self.assertNotEqual(row, None) 235 self.assertNotEqual(row, None)
246 self.assertEqual(row[0], 1) 236 self.assertEqual(row[0], 1)
247 237
248 def _exec(self): 238 def _exec(self):
249 self.cursor.execute(self.sql) 239 self.cursor.execute(self.sql)
250 240
251 def test_close_cnxn(self): 241 def test_close_cnxn(self):
252 """Make sure using a Cursor after closing its connection doesn't crash.""" 242 """Make sure using a Cursor after closing its connection doesn't crash."""
253 243
254 self.cursor.execute("create table t1(id integer, s varchar(20))") 244 self.cursor.execute("create table t1(id integer, s varchar(20))")
255 self.cursor.execute("insert into t1 values (?,?)", 1, 'test') 245 self.cursor.execute("insert into t1 values (?,?)", 1, 'test')
256 self.cursor.execute("select * from t1") 246 self.cursor.execute("select * from t1")
257 247
258 self.cnxn.close() 248 self.cnxn.close()
259 249
260 # Now that the connection is closed, we expect an exception. (If the code attempts to use 250 # Now that the connection is closed, we expect an exception. (If the code attempts to use
261 # the HSTMT, we'll get an access violation instead.) 251 # the HSTMT, we'll get an access violation instead.)
262 self.sql = "select * from t1" 252 self.sql = "select * from t1"
263 self.assertRaises(pyodbc.ProgrammingError, self._exec) 253 self.assertRaises(pyodbc.ProgrammingError, self._exec)
264 254
265 def test_empty_unicode(self):
266 self.cursor.execute("create table t1(s nvarchar(20))")
267 self.cursor.execute("insert into t1 values(?)", u"")
268
269 def test_unicode_query(self):
270 self.cursor.execute(u"select 1")
271
272 def test_negative_row_index(self): 255 def test_negative_row_index(self):
273 self.cursor.execute("create table t1(s varchar(20))") 256 self.cursor.execute("create table t1(s varchar(20))")
274 self.cursor.execute("insert into t1 values(?)", "1") 257 self.cursor.execute("insert into t1 values(?)", "1")
275 row = self.cursor.execute("select * from t1").fetchone() 258 row = self.cursor.execute("select * from t1").fetchone()
276 self.assertEquals(row[0], "1") 259 self.assertEquals(row[0], "1")
277 self.assertEquals(row[-1], "1") 260 self.assertEquals(row[-1], "1")
278 261
279 def test_version(self): 262 def test_version(self):
280 self.assertEquals(3, len(pyodbc.version.split('.'))) # 1.3.1 etc. 263 self.assertEquals(3, len(pyodbc.version.split('.'))) # 1.3.1 etc.
281 264
282 # 265 #
283 # ints and floats 266 # ints and floats
284 # 267 #
285 268
286 def test_int(self): 269 def test_int(self):
287 value = 1234 270 value = 1234
288 self.cursor.execute("create table t1(n int)") 271 self.cursor.execute("create table t1(n int)")
289 self.cursor.execute("insert into t1 values (?)", value) 272 self.cursor.execute("insert into t1 values (?)", value)
290 result = self.cursor.execute("select n from t1").fetchone()[0] 273 result = self.cursor.execute("select n from t1").fetchone()[0]
291 self.assertEquals(result, value) 274 self.assertEquals(result, value)
292 275
293 def test_negative_int(self): 276 def test_negative_int(self):
294 value = -1 277 value = -1
295 self.cursor.execute("create table t1(n int)") 278 self.cursor.execute("create table t1(n int)")
296 self.cursor.execute("insert into t1 values (?)", value) 279 self.cursor.execute("insert into t1 values (?)", value)
297 result = self.cursor.execute("select n from t1").fetchone()[0] 280 result = self.cursor.execute("select n from t1").fetchone()[0]
298 self.assertEquals(result, value) 281 self.assertEquals(result, value)
299 282
300 def test_bigint(self): 283 def test_bigint(self):
301 input = 3000000000 284 input = 3000000000
302 self.cursor.execute("create table t1(d bigint)") 285 self.cursor.execute("create table t1(d bigint)")
303 self.cursor.execute("insert into t1 values (?)", input) 286 self.cursor.execute("insert into t1 values (?)", input)
304 result = self.cursor.execute("select d from t1").fetchone()[0] 287 result = self.cursor.execute("select d from t1").fetchone()[0]
305 self.assertEqual(result, input) 288 self.assertEqual(result, input)
306 289
307 def test_negative_bigint(self): 290 def test_negative_bigint(self):
308 # Issue 186: BIGINT problem on 32-bit architeture 291 # Issue 186: BIGINT problem on 32-bit architeture
309 input = -430000000 292 input = -430000000
310 self.cursor.execute("create table t1(d bigint)") 293 self.cursor.execute("create table t1(d bigint)")
311 self.cursor.execute("insert into t1 values (?)", input) 294 self.cursor.execute("insert into t1 values (?)", input)
312 result = self.cursor.execute("select d from t1").fetchone()[0] 295 result = self.cursor.execute("select d from t1").fetchone()[0]
313 self.assertEqual(result, input) 296 self.assertEqual(result, input)
314 297
315 def test_float(self): 298 def test_float(self):
316 value = 1234.567 299 value = 1234.567
317 self.cursor.execute("create table t1(n float)") 300 self.cursor.execute("create table t1(n float)")
318 self.cursor.execute("insert into t1 values (?)", value) 301 self.cursor.execute("insert into t1 values (?)", value)
319 result = self.cursor.execute("select n from t1").fetchone()[0] 302 result = self.cursor.execute("select n from t1").fetchone()[0]
320 self.assertEquals(result, value) 303 self.assertEquals(result, value)
321 304
322 def test_negative_float(self): 305 def test_negative_float(self):
323 value = -200 306 value = -200
324 self.cursor.execute("create table t1(n float)") 307 self.cursor.execute("create table t1(n float)")
325 self.cursor.execute("insert into t1 values (?)", value) 308 self.cursor.execute("insert into t1 values (?)", value)
326 result = self.cursor.execute("select n from t1").fetchone()[0] 309 result = self.cursor.execute("select n from t1").fetchone()[0]
327 self.assertEqual(value, result) 310 self.assertEqual(value, result)
328 311
329 # 312 #
330 # rowcount 313 # rowcount
331 # 314 #
332 315
333 def test_rowcount_delete(self): 316 def test_rowcount_delete(self):
334 self.assertEquals(self.cursor.rowcount, -1) 317 self.assertEquals(self.cursor.rowcount, -1)
335 self.cursor.execute("create table t1(i int)") 318 self.cursor.execute("create table t1(i int)")
336 count = 4 319 count = 4
337 for i in range(count): 320 for i in range(count):
338 self.cursor.execute("insert into t1 values (?)", i) 321 self.cursor.execute("insert into t1 values (?)", i)
339 self.cursor.execute("delete from t1") 322 self.cursor.execute("delete from t1")
340 self.assertEquals(self.cursor.rowcount, count) 323 self.assertEquals(self.cursor.rowcount, count)
341 324
342 def test_rowcount_nodata(self): 325 def test_rowcount_nodata(self):
343 """ 326 """
344 This represents a different code path than a delete that deleted something. 327 This represents a different code path than a delete that deleted something.
345 328
346 The return value is SQL_NO_DATA and code after it was causing an error. We could use SQL_NO_DATA to step over 329 The return value is SQL_NO_DATA and code after it was causing an error. We could use SQL_NO_DATA to step over
347 the code that errors out and drop down to the same SQLRowCount code. On the other hand, we could hardcode a 330 the code that errors out and drop down to the same SQLRowCount code. On the other hand, we could hardcode a
348 zero return value. 331 zero return value.
349 """ 332 """
350 self.cursor.execute("create table t1(i int)") 333 self.cursor.execute("create table t1(i int)")
351 # This is a different code path internally. 334 # This is a different code path internally.
352 self.cursor.execute("delete from t1") 335 self.cursor.execute("delete from t1")
353 self.assertEquals(self.cursor.rowcount, 0) 336 self.assertEquals(self.cursor.rowcount, 0)
354 337
355 def test_rowcount_select(self): 338 def test_rowcount_select(self):
356 """ 339 """
357 Ensure Cursor.rowcount is set properly after a select statement. 340 Ensure Cursor.rowcount is set properly after a select statement.
358 """ 341 """
359 self.cursor.execute("create table t1(i int)") 342 self.cursor.execute("create table t1(i int)")
360 count = 4 343 count = 4
361 for i in range(count): 344 for i in range(count):
362 self.cursor.execute("insert into t1 values (?)", i) 345 self.cursor.execute("insert into t1 values (?)", i)
363 self.cursor.execute("select * from t1") 346 self.cursor.execute("select * from t1")
364 self.assertEquals(self.cursor.rowcount, count) 347 self.assertEquals(self.cursor.rowcount, count)
365 348
366 rows = self.cursor.fetchall() 349 rows = self.cursor.fetchall()
367 self.assertEquals(len(rows), count) 350 self.assertEquals(len(rows), count)
368 self.assertEquals(self.cursor.rowcount, count) 351 self.assertEquals(self.cursor.rowcount, count)
369 352
370 # Fails. Not terribly important so I'm going to comment out for now and report to the ODBC driver writer. 353 # Fails. Not terribly important so I'm going to comment out for now and report to the ODBC driver writer.
371 # def test_rowcount_reset(self): 354 # def test_rowcount_reset(self):
372 # "Ensure rowcount is reset to -1" 355 # "Ensure rowcount is reset to -1"
373 # self.cursor.execute("create table t1(i int)") 356 # self.cursor.execute("create table t1(i int)")
374 # count = 4 357 # count = 4
375 # for i in range(count): 358 # for i in range(count):
376 # self.cursor.execute("insert into t1 values (?)", i) 359 # self.cursor.execute("insert into t1 values (?)", i)
377 # self.assertEquals(self.cursor.rowcount, 1) 360 # self.assertEquals(self.cursor.rowcount, 1)
378 # 361 #
379 # self.cursor.execute("create table t2(i int)") 362 # self.cursor.execute("create table t2(i int)")
380 # self.assertEquals(self.cursor.rowcount, -1) 363 # self.assertEquals(self.cursor.rowcount, -1)
381 364
382 # 365 #
383 # always return Cursor 366 # always return Cursor
384 # 367 #
385 368
386 # In the 2.0.x branch, Cursor.execute sometimes returned the cursor and sometimes the rowcount. This proved very 369 # In the 2.0.x branch, Cursor.execute sometimes returned the cursor and sometimes the rowcount. This proved very
387 # confusing when things went wrong and added very little value even when things went right since users could always 370 # confusing when things went wrong and added very little value even when things went right since users could always
388 # use: cursor.execute("...").rowcount 371 # use: cursor.execute("...").rowcount
389 372
390 def test_retcursor_delete(self): 373 def test_retcursor_delete(self):
391 self.cursor.execute("create table t1(i int)") 374 self.cursor.execute("create table t1(i int)")
392 self.cursor.execute("insert into t1 values (1)") 375 self.cursor.execute("insert into t1 values (1)")
393 v = self.cursor.execute("delete from t1") 376 v = self.cursor.execute("delete from t1")
394 self.assertEquals(v, self.cursor) 377 self.assertEquals(v, self.cursor)
395 378
396 def test_retcursor_nodata(self): 379 def test_retcursor_nodata(self):
397 """ 380 """
398 This represents a different code path than a delete that deleted something. 381 This represents a different code path than a delete that deleted something.
399 382
400 The return value is SQL_NO_DATA and code after it was causing an error. We could use SQL_NO_DATA to step over 383 The return value is SQL_NO_DATA and code after it was causing an error. We could use SQL_NO_DATA to step over
401 the code that errors out and drop down to the same SQLRowCount code. 384 the code that errors out and drop down to the same SQLRowCount code.
402 """ 385 """
403 self.cursor.execute("create table t1(i int)") 386 self.cursor.execute("create table t1(i int)")
404 # This is a different code path internally. 387 # This is a different code path internally.
405 v = self.cursor.execute("delete from t1") 388 v = self.cursor.execute("delete from t1")
406 self.assertEquals(v, self.cursor) 389 self.assertEquals(v, self.cursor)
407 390
408 def test_retcursor_select(self): 391 def test_retcursor_select(self):
409 self.cursor.execute("create table t1(i int)") 392 self.cursor.execute("create table t1(i int)")
410 self.cursor.execute("insert into t1 values (1)") 393 self.cursor.execute("insert into t1 values (1)")
411 v = self.cursor.execute("select * from t1") 394 v = self.cursor.execute("select * from t1")
412 self.assertEquals(v, self.cursor) 395 self.assertEquals(v, self.cursor)
413 396
414 # 397 #
415 # misc 398 # misc
416 # 399 #
417 400
418 def test_lower_case(self): 401 def test_lower_case(self):
419 "Ensure pyodbc.lowercase forces returned column names to lowercase." 402 "Ensure pyodbc.lowercase forces returned column names to lowercase."
420 403
421 # Has to be set before creating the cursor, so we must recreate self.cursor. 404 # Has to be set before creating the cursor, so we must recreate self.cursor.
422 405
423 pyodbc.lowercase = True 406 pyodbc.lowercase = True
424 self.cursor = self.cnxn.cursor() 407 self.cursor = self.cnxn.cursor()
425 408
426 self.cursor.execute("create table t1(Abc int, dEf int)") 409 self.cursor.execute("create table t1(Abc int, dEf int)")
427 self.cursor.execute("select * from t1") 410 self.cursor.execute("select * from t1")
428 411
429 names = [ t[0] for t in self.cursor.description ] 412 names = [ t[0] for t in self.cursor.description ]
430 names.sort() 413 names.sort()
431 414
432 self.assertEquals(names, [ "abc", "def" ]) 415 self.assertEquals(names, [ "abc", "def" ])
433 416
434 # Put it back so other tests don't fail. 417 # Put it back so other tests don't fail.
435 pyodbc.lowercase = False 418 pyodbc.lowercase = False
436 419
437 def test_row_description(self): 420 def test_row_description(self):
438 """ 421 """
439 Ensure Cursor.description is accessible as Row.cursor_description. 422 Ensure Cursor.description is accessible as Row.cursor_description.
440 """ 423 """
441 self.cursor = self.cnxn.cursor() 424 self.cursor = self.cnxn.cursor()
442 self.cursor.execute("create table t1(a int, b char(3))") 425 self.cursor.execute("create table t1(a int, b char(3))")
443 self.cnxn.commit() 426 self.cnxn.commit()
444 self.cursor.execute("insert into t1 values(1, 'abc')") 427 self.cursor.execute("insert into t1 values(1, 'abc')")
445 428
446 row = self.cursor.execute("select * from t1").fetchone() 429 row = self.cursor.execute("select * from t1").fetchone()
447 430
448 self.assertEquals(self.cursor.description, row.cursor_description) 431 self.assertEquals(self.cursor.description, row.cursor_description)
449 432
450 433
451 def test_executemany(self): 434 def test_executemany(self):
452 self.cursor.execute("create table t1(a int, b varchar(10))") 435 self.cursor.execute("create table t1(a int, b varchar(10))")
453 436
454 params = [ (i, str(i)) for i in range(1, 6) ] 437 params = [ (i, str(i)) for i in range(1, 6) ]
455 438
456 self.cursor.executemany("insert into t1(a, b) values (?,?)", params) 439 self.cursor.executemany("insert into t1(a, b) values (?,?)", params)
457 440
458 count = self.cursor.execute("select count(*) from t1").fetchone()[0] 441 count = self.cursor.execute("select count(*) from t1").fetchone()[0]
459 self.assertEqual(count, len(params)) 442 self.assertEqual(count, len(params))
460 443
461 self.cursor.execute("select a, b from t1 order by a") 444 self.cursor.execute("select a, b from t1 order by a")
462 rows = self.cursor.fetchall() 445 rows = self.cursor.fetchall()
463 self.assertEqual(count, len(rows)) 446 self.assertEqual(count, len(rows))
464 447
465 for param, row in zip(params, rows): 448 for param, row in zip(params, rows):
466 self.assertEqual(param[0], row[0]) 449 self.assertEqual(param[0], row[0])
467 self.assertEqual(param[1], row[1]) 450 self.assertEqual(param[1], row[1])
468 451
469 452
470 def test_executemany_one(self): 453 def test_executemany_one(self):
471 "Pass executemany a single sequence" 454 "Pass executemany a single sequence"
472 self.cursor.execute("create table t1(a int, b varchar(10))") 455 self.cursor.execute("create table t1(a int, b varchar(10))")
473 456
474 params = [ (1, "test") ] 457 params = [ (1, "test") ]
475 458
476 self.cursor.executemany("insert into t1(a, b) values (?,?)", params) 459 self.cursor.executemany("insert into t1(a, b) values (?,?)", params)
477 460
478 count = self.cursor.execute("select count(*) from t1").fetchone()[0] 461 count = self.cursor.execute("select count(*) from t1").fetchone()[0]
479 self.assertEqual(count, len(params)) 462 self.assertEqual(count, len(params))
480 463
481 self.cursor.execute("select a, b from t1 order by a") 464 self.cursor.execute("select a, b from t1 order by a")
482 rows = self.cursor.fetchall() 465 rows = self.cursor.fetchall()
483 self.assertEqual(count, len(rows)) 466 self.assertEqual(count, len(rows))
484 467
485 for param, row in zip(params, rows): 468 for param, row in zip(params, rows):
486 self.assertEqual(param[0], row[0]) 469 self.assertEqual(param[0], row[0])
487 self.assertEqual(param[1], row[1]) 470 self.assertEqual(param[1], row[1])
488 471
489 472
490 def test_executemany_failure(self): 473 def test_executemany_failure(self):
491 """ 474 """
492 Ensure that an exception is raised if one query in an executemany fails. 475 Ensure that an exception is raised if one query in an executemany fails.
493 """ 476 """
494 self.cursor.execute("create table t1(a int, b varchar(10))") 477 self.cursor.execute("create table t1(a int, b varchar(10))")
495 478
496 params = [ (1, 'good'), 479 params = [ (1, 'good'),
497 ('error', 'not an int'), 480 ('error', 'not an int'),
498 (3, 'good') ] 481 (3, 'good') ]
499 482
500 self.failUnlessRaises(pyodbc.Error, self.cursor.executemany, "insert into t1(a, b) value (?, ?)", params) 483 self.failUnlessRaises(pyodbc.Error, self.cursor.executemany, "insert into t1(a, b) value (?, ?)", params)
501 484
502 485
503 def test_row_slicing(self): 486 def test_row_slicing(self):
504 self.cursor.execute("create table t1(a int, b int, c int, d int)"); 487 self.cursor.execute("create table t1(a int, b int, c int, d int)");
505 self.cursor.execute("insert into t1 values(1,2,3,4)") 488 self.cursor.execute("insert into t1 values(1,2,3,4)")
506 489
507 row = self.cursor.execute("select * from t1").fetchone() 490 row = self.cursor.execute("select * from t1").fetchone()
508 491
509 result = row[:] 492 result = row[:]
510 self.failUnless(result is row) 493 self.failUnless(result is row)
511 494
512 result = row[:-1] 495 result = row[:-1]
513 self.assertEqual(result, (1,2,3)) 496 self.assertEqual(result, (1,2,3))
514 497
515 result = row[0:4] 498 result = row[0:4]
516 self.failUnless(result is row) 499 self.failUnless(result is row)
517 500
518 501
519 def test_row_repr(self): 502 def test_row_repr(self):
520 self.cursor.execute("create table t1(a int, b int, c int, d int)"); 503 self.cursor.execute("create table t1(a int, b int, c int, d int)");
521 self.cursor.execute("insert into t1 values(1,2,3,4)") 504 self.cursor.execute("insert into t1 values(1,2,3,4)")
522 505
523 row = self.cursor.execute("select * from t1").fetchone() 506 row = self.cursor.execute("select * from t1").fetchone()
524 507
525 result = str(row) 508 result = str(row)
526 self.assertEqual(result, "(1, 2, 3, 4)") 509 self.assertEqual(result, "(1, 2, 3, 4)")
527 510
528 result = str(row[:-1]) 511 result = str(row[:-1])
529 self.assertEqual(result, "(1, 2, 3)") 512 self.assertEqual(result, "(1, 2, 3)")
530 513
531 result = str(row[:1]) 514 result = str(row[:1])
532 self.assertEqual(result, "(1,)") 515 self.assertEqual(result, "(1,)")
533 516
534 517
535 def test_view_select(self): 518 def test_view_select(self):
536 # Reported in forum: Can't select from a view? I think I do this a lot, but another test never hurts. 519 # Reported in forum: Can't select from a view? I think I do this a lot, but another test never hurts.
537 520
538 # Create a table (t1) with 3 rows and a view (t2) into it. 521 # Create a table (t1) with 3 rows and a view (t2) into it.
539 self.cursor.execute("create table t1(c1 int identity(1, 1), c2 varchar(50))") 522 self.cursor.execute("create table t1(c1 int identity(1, 1), c2 varchar(50))")
540 for i in range(3): 523 for i in range(3):
541 self.cursor.execute("insert into t1(c2) values (?)", "string%s" % i) 524 self.cursor.execute("insert into t1(c2) values (?)", "string%s" % i)
542 self.cursor.execute("create view t2 as select * from t1") 525 self.cursor.execute("create view t2 as select * from t1")
543 526
544 # Select from the view 527 # Select from the view
545 self.cursor.execute("select * from t2") 528 self.cursor.execute("select * from t2")
546 rows = self.cursor.fetchall() 529 rows = self.cursor.fetchall()
547 self.assert_(rows is not None) 530 self.assert_(rows is not None)
548 self.assert_(len(rows) == 3) 531 self.assert_(len(rows) == 3)
549 532
550 def test_autocommit(self): 533 def test_autocommit(self):
551 self.assertEqual(self.cnxn.autocommit, False) 534 self.assertEqual(self.cnxn.autocommit, False)
552 535
553 othercnxn = pyodbc.connect(self.connection_string, autocommit=True) 536 othercnxn = pyodbc.connect(self.connection_string, autocommit=True)
554 self.assertEqual(othercnxn.autocommit, True) 537 self.assertEqual(othercnxn.autocommit, True)
555 538
556 othercnxn.autocommit = False 539 othercnxn.autocommit = False
557 self.assertEqual(othercnxn.autocommit, False) 540 self.assertEqual(othercnxn.autocommit, False)
558 541
559 def test_unicode_results(self):
560 "Ensure unicode_results forces Unicode"
561 othercnxn = pyodbc.connect(self.connection_string, unicode_results=True)
562 othercursor = othercnxn.cursor()
563
564 # ANSI data in an ANSI column ...
565 othercursor.execute("create table t1(s varchar(20))")
566 othercursor.execute("insert into t1 values(?)", 'test')
567
568 # ... should be returned as Unicode
569 value = othercursor.execute("select s from t1").fetchone()[0]
570 self.assertEqual(value, u'test')
571
572 def test_skip(self): 542 def test_skip(self):
573 # Insert 1, 2, and 3. Fetch 1, skip 2, fetch 3. 543 # Insert 1, 2, and 3. Fetch 1, skip 2, fetch 3.
574 544
575 self.cursor.execute("create table t1(id int)"); 545 self.cursor.execute("create table t1(id int)");
576 for i in range(1, 5): 546 for i in range(1, 5):
577 self.cursor.execute("insert into t1 values(?)", i) 547 self.cursor.execute("insert into t1 values(?)", i)
578 self.cursor.execute("select id from t1 order by id") 548 self.cursor.execute("select id from t1 order by id")
579 self.assertEqual(self.cursor.fetchone()[0], 1) 549 self.assertEqual(self.cursor.fetchone()[0], 1)
580 self.cursor.skip(2) 550 self.cursor.skip(2)
581 self.assertEqual(self.cursor.fetchone()[0], 4) 551 self.assertEqual(self.cursor.fetchone()[0], 4)
582 552
583 def test_sets_execute(self): 553 def test_sets_execute(self):
584 # Only lists and tuples are allowed. 554 # Only lists and tuples are allowed.
585 def f(): 555 def f():
586 self.cursor.execute("create table t1 (word varchar (100))") 556 self.cursor.execute("create table t1 (word varchar (100))")
587 words = set (['a']) 557 words = set (['a'])
588 self.cursor.execute("insert into t1 (word) VALUES (?)", [words]) 558 self.cursor.execute("insert into t1 (word) VALUES (?)", [words])
589 559
590 self.assertRaises(pyodbc.ProgrammingError, f) 560 self.assertRaises(pyodbc.ProgrammingError, f)
591 561
592 def test_sets_executemany(self): 562 def test_sets_executemany(self):
593 # Only lists and tuples are allowed. 563 # Only lists and tuples are allowed.
594 def f(): 564 def f():
595 self.cursor.execute("create table t1 (word varchar (100))") 565 self.cursor.execute("create table t1 (word varchar (100))")
596 words = set (['a']) 566 words = set (['a'])
597 self.cursor.executemany("insert into t1 (word) values (?)", [words]) 567 self.cursor.executemany("insert into t1 (word) values (?)", [words])
598 568
599 self.assertRaises(TypeError, f) 569 self.assertRaises(TypeError, f)
600 570
601 def test_row_execute(self): 571 def test_row_execute(self):
602 "Ensure we can use a Row object as a parameter to execute" 572 "Ensure we can use a Row object as a parameter to execute"
603 self.cursor.execute("create table t1(n int, s varchar(10))") 573 self.cursor.execute("create table t1(n int, s varchar(10))")
604 self.cursor.execute("insert into t1 values (1, 'a')") 574 self.cursor.execute("insert into t1 values (1, 'a')")
605 row = self.cursor.execute("select n, s from t1").fetchone() 575 row = self.cursor.execute("select n, s from t1").fetchone()
606 self.assertNotEqual(row, None) 576 self.assertNotEqual(row, None)
607 577
608 self.cursor.execute("create table t2(n int, s varchar(10))") 578 self.cursor.execute("create table t2(n int, s varchar(10))")
609 self.cursor.execute("insert into t2 values (?, ?)", row) 579 self.cursor.execute("insert into t2 values (?, ?)", row)
610 580
611 def test_row_executemany(self): 581 def test_row_executemany(self):
612 "Ensure we can use a Row object as a parameter to executemany" 582 "Ensure we can use a Row object as a parameter to executemany"
613 self.cursor.execute("create table t1(n int, s varchar(10))") 583 self.cursor.execute("create table t1(n int, s varchar(10))")
614 584
615 for i in range(3): 585 for i in range(3):
616 self.cursor.execute("insert into t1 values (?, ?)", i, chr(ord('a')+i)) 586 self.cursor.execute("insert into t1 values (?, ?)", i, chr(ord('a')+i))
617 587
618 rows = self.cursor.execute("select n, s from t1").fetchall() 588 rows = self.cursor.execute("select n, s from t1").fetchall()
619 self.assertNotEqual(len(rows), 0) 589 self.assertNotEqual(len(rows), 0)
620 590
621 self.cursor.execute("create table t2(n int, s varchar(10))") 591 self.cursor.execute("create table t2(n int, s varchar(10))")
622 self.cursor.executemany("insert into t2 values (?, ?)", rows) 592 self.cursor.executemany("insert into t2 values (?, ?)", rows)
623 593
624 def test_description(self): 594 def test_description(self):
625 "Ensure cursor.description is correct" 595 "Ensure cursor.description is correct"
626 596
627 self.cursor.execute("create table t1(n int, s text)") 597 self.cursor.execute("create table t1(n int, s text)")
628 self.cursor.execute("insert into t1 values (1, 'abc')") 598 self.cursor.execute("insert into t1 values (1, 'abc')")
629 self.cursor.execute("select * from t1") 599 self.cursor.execute("select * from t1")
630 600
631 # (I'm not sure the precision of an int is constant across different versions, bits, so I'm hand checking the 601 # (I'm not sure the precision of an int is constant across different versions, bits, so I'm hand checking the
632 # items I do know. 602 # items I do know.
633 603
634 # int 604 # int
635 t = self.cursor.description[0] 605 t = self.cursor.description[0]
636 self.assertEqual(t[0], 'n') 606 self.assertEqual(t[0], 'n')
637 self.assertEqual(t[1], int) 607 self.assertEqual(t[1], int)
638 self.assertEqual(t[5], 0) # scale 608 self.assertEqual(t[5], 0) # scale
639 self.assertEqual(t[6], True) # nullable 609 self.assertEqual(t[6], True) # nullable
640 610
641 # text 611 # text
642 t = self.cursor.description[1] 612 t = self.cursor.description[1]
643 self.assertEqual(t[0], 's') 613 self.assertEqual(t[0], 's')
644 self.assertEqual(t[1], unicode) 614 self.assertEqual(t[1], str)
645 self.assertEqual(t[5], 0) # scale 615 self.assertEqual(t[5], 0) # scale
646 self.assertEqual(t[6], True) # nullable 616 self.assertEqual(t[6], True) # nullable
647 617
648 def test_none_param(self):
649 "Ensure None can be used for params other than the first"
650 # Some driver/db versions would fail if NULL was not the first parameter because SQLDescribeParam (only used
651 # with NULL) could not be used after the first call to SQLBindParameter. This means None always worked for the
652 # first column, but did not work for later columns.
653 #
654 # If SQLDescribeParam doesn't work, pyodbc would use VARCHAR which almost always worked. However,
655 # binary/varbinary won't allow an implicit conversion.
656
657 value = u'\x12abc'
658 self.cursor.execute("create table t1(n int, b blob)")
659 self.cursor.execute("insert into t1 values (1, ?)", value)
660 row = self.cursor.execute("select * from t1").fetchone()
661 self.assertEqual(row.n, 1)
662 self.assertEqual(type(row.b), buffer)
663 self.assertEqual(row.b, value)
664
665
666 def test_row_equal(self): 618 def test_row_equal(self):
667 self.cursor.execute("create table t1(n int, s varchar(20))") 619 self.cursor.execute("create table t1(n int, s varchar(20))")
668 self.cursor.execute("insert into t1 values (1, 'test')") 620 self.cursor.execute("insert into t1 values (1, 'test')")
669 row1 = self.cursor.execute("select n, s from t1").fetchone() 621 row1 = self.cursor.execute("select n, s from t1").fetchone()
670 row2 = self.cursor.execute("select n, s from t1").fetchone() 622 row2 = self.cursor.execute("select n, s from t1").fetchone()
671 b = (row1 == row2) 623 b = (row1 == row2)
672 self.assertEqual(b, True) 624 self.assertEqual(b, True)
673 625
674 def test_row_gtlt(self): 626 def test_row_gtlt(self):
675 self.cursor.execute("create table t1(n int, s varchar(20))") 627 self.cursor.execute("create table t1(n int, s varchar(20))")
676 self.cursor.execute("insert into t1 values (1, 'test1')") 628 self.cursor.execute("insert into t1 values (1, 'test1')")
677 self.cursor.execute("insert into t1 values (1, 'test2')") 629 self.cursor.execute("insert into t1 values (1, 'test2')")
678 rows = self.cursor.execute("select n, s from t1 order by s").fetchall() 630 rows = self.cursor.execute("select n, s from t1 order by s").fetchall()
679 self.assert_(rows[0] < rows[1]) 631 self.assert_(rows[0] < rows[1])
680 self.assert_(rows[0] <= rows[1]) 632 self.assert_(rows[0] <= rows[1])
681 self.assert_(rows[1] > rows[0]) 633 self.assert_(rows[1] > rows[0])
682 self.assert_(rows[1] >= rows[0]) 634 self.assert_(rows[1] >= rows[0])
683 self.assert_(rows[0] != rows[1]) 635 self.assert_(rows[0] != rows[1])
684 636
685 rows = list(rows) 637 rows = list(rows)
686 rows.sort() # uses < 638 rows.sort() # uses <
687 639
688 def test_context_manager(self): 640 def test_context_manager(self):
689 with pyodbc.connect(self.connection_string) as cnxn: 641 with pyodbc.connect(self.connection_string) as cnxn:
690 cnxn.getinfo(pyodbc.SQL_DEFAULT_TXN_ISOLATION) 642 cnxn.getinfo(pyodbc.SQL_DEFAULT_TXN_ISOLATION)
691 643
692 # The connection should be closed now. 644 # The connection should be closed now.
693 def test(): 645 def test():
694 cnxn.getinfo(pyodbc.SQL_DEFAULT_TXN_ISOLATION) 646 cnxn.getinfo(pyodbc.SQL_DEFAULT_TXN_ISOLATION)
695 self.assertRaises(pyodbc.ProgrammingError, test) 647 self.assertRaises(pyodbc.ProgrammingError, test)
696 648
697 def test_untyped_none(self): 649 def test_untyped_none(self):
698 # From issue 129 650 # From issue 129
699 value = self.cursor.execute("select ?", None).fetchone()[0] 651 value = self.cursor.execute("select ?", None).fetchone()[0]
700 self.assertEqual(value, None) 652 self.assertEqual(value, None)
701 653
702 def test_large_update_nodata(self): 654 def test_large_update_nodata(self):
703 self.cursor.execute('create table t1(a blob)') 655 self.cursor.execute('create table t1(a blob)')
704 hundredkb = buffer('x'*100*1024) 656 hundredkb = bytearray('x'*100*1024)
705 self.cursor.execute('update t1 set a=? where 1=0', (hundredkb,)) 657 self.cursor.execute('update t1 set a=? where 1=0', (hundredkb,))
706 658
707 def test_no_fetch(self): 659 def test_no_fetch(self):
708 # Issue 89 with FreeTDS: Multiple selects (or catalog functions that issue selects) without fetches seem to 660 # Issue 89 with FreeTDS: Multiple selects (or catalog functions that issue selects) without fetches seem to
709 # confuse the driver. 661 # confuse the driver.
710 self.cursor.execute('select 1') 662 self.cursor.execute('select 1')
711 self.cursor.execute('select 1') 663 self.cursor.execute('select 1')
712 self.cursor.execute('select 1') 664 self.cursor.execute('select 1')
713 665
714 def main(): 666 def main():
715 from optparse import OptionParser 667 from optparse import OptionParser
716 parser = OptionParser(usage=usage) 668 parser = OptionParser(usage=usage)
717 parser.add_option("-v", "--verbose", action="count", help="Increment test verbosity (can be used multiple times)") 669 parser.add_option("-v", "--verbose", action="count", help="Increment test verbosity (can be used multiple times)")
718 parser.add_option("-d", "--debug", action="store_true", default=False, help="Print debugging items") 670 parser.add_option("-d", "--debug", action="store_true", default=False, help="Print debugging items")
719 parser.add_option("-t", "--test", help="Run only the named test") 671 parser.add_option("-t", "--test", help="Run only the named test")
720 672
721 (options, args) = parser.parse_args() 673 (options, args) = parser.parse_args()
722 674
723 if len(args) > 1: 675 if len(args) > 1:
724 parser.error('Only one argument is allowed. Do you need quotes around the connection string?') 676 parser.error('Only one argument is allowed. Do you need quotes around the connection string?')
725 677
726 if not args: 678 if not args:
727 connection_string = load_setup_connection_string('sqlitetests') 679 connection_string = load_setup_connection_string('sqlitetests')
728 print 'connection_string:', connection_string
729 680
730 if not connection_string: 681 if not connection_string:
731 parser.print_help() 682 parser.print_help()
732 raise SystemExit() 683 raise SystemExit()
733 else: 684 else:
734 connection_string = args[0] 685 connection_string = args[0]
735 686
736 cnxn = pyodbc.connect(connection_string) 687 cnxn = pyodbc.connect(connection_string)
737 print_library_info(cnxn) 688 print_library_info(cnxn)
738 cnxn.close() 689 cnxn.close()
739 690
740 suite = load_tests(SqliteTestCase, options.test, connection_string) 691 suite = load_tests(SqliteTestCase, options.test, connection_string)
741 692
742 testRunner = unittest.TextTestRunner(verbosity=options.verbose) 693 testRunner = unittest.TextTestRunner(verbosity=options.verbose)
743 result = testRunner.run(suite) 694 result = testRunner.run(suite)
744 695
745 696
746 if __name__ == '__main__': 697 if __name__ == '__main__':
747 698
748 # Add the build directory to the path so we're testing the latest build, not the installed version. 699 # Add the build directory to the path so we're testing the latest build, not the installed version.
749 700
750 add_to_path() 701 add_to_path()
751 702
752 import pyodbc 703 import pyodbc
753 main() 704 main()
Powered by Google Project Hosting