Python code coverage for Lib/sqlite3/test/dbapi.py

#	count	content
1	n/a	#-- coding: iso-8859-1 --
2	n/a	# pysqlite2/test/dbapi.py: tests for DB-API compliance
3	n/a	#
4	n/a	# Copyright (C) 2004-2010 Gerhard Häring <gh@ghaering.de>
5	n/a	#
6	n/a	# This file is part of pysqlite.
7	n/a	#
8	n/a	# This software is provided 'as-is', without any express or implied
9	n/a	# warranty. In no event will the authors be held liable for any damages
10	n/a	# arising from the use of this software.
11	n/a	#
12	n/a	# Permission is granted to anyone to use this software for any purpose,
13	n/a	# including commercial applications, and to alter it and redistribute it
14	n/a	# freely, subject to the following restrictions:
15	n/a	#
16	n/a	# 1. The origin of this software must not be misrepresented; you must not
17	n/a	# claim that you wrote the original software. If you use this software
18	n/a	# in a product, an acknowledgment in the product documentation would be
19	n/a	# appreciated but is not required.
20	n/a	# 2. Altered source versions must be plainly marked as such, and must not be
21	n/a	# misrepresented as being the original software.
22	n/a	# 3. This notice may not be removed or altered from any source distribution.
23	n/a
24	n/a	import unittest
25	n/a	import sqlite3 as sqlite
26	n/a	try:
27	n/a	import threading
28	n/a	except ImportError:
29	n/a	threading = None
30	n/a
31	n/a	from test.support import TESTFN, unlink
32	n/a
33	n/a
34	n/a	class ModuleTests(unittest.TestCase):
35	n/a	def CheckAPILevel(self):
36	n/a	self.assertEqual(sqlite.apilevel, "2.0",
37	n/a	"apilevel is %s, should be 2.0" % sqlite.apilevel)
38	n/a
39	n/a	def CheckThreadSafety(self):
40	n/a	self.assertEqual(sqlite.threadsafety, 1,
41	n/a	"threadsafety is %d, should be 1" % sqlite.threadsafety)
42	n/a
43	n/a	def CheckParamStyle(self):
44	n/a	self.assertEqual(sqlite.paramstyle, "qmark",
45	n/a	"paramstyle is '%s', should be 'qmark'" %
46	n/a	sqlite.paramstyle)
47	n/a
48	n/a	def CheckWarning(self):
49	n/a	self.assertTrue(issubclass(sqlite.Warning, Exception),
50	n/a	"Warning is not a subclass of Exception")
51	n/a
52	n/a	def CheckError(self):
53	n/a	self.assertTrue(issubclass(sqlite.Error, Exception),
54	n/a	"Error is not a subclass of Exception")
55	n/a
56	n/a	def CheckInterfaceError(self):
57	n/a	self.assertTrue(issubclass(sqlite.InterfaceError, sqlite.Error),
58	n/a	"InterfaceError is not a subclass of Error")
59	n/a
60	n/a	def CheckDatabaseError(self):
61	n/a	self.assertTrue(issubclass(sqlite.DatabaseError, sqlite.Error),
62	n/a	"DatabaseError is not a subclass of Error")
63	n/a
64	n/a	def CheckDataError(self):
65	n/a	self.assertTrue(issubclass(sqlite.DataError, sqlite.DatabaseError),
66	n/a	"DataError is not a subclass of DatabaseError")
67	n/a
68	n/a	def CheckOperationalError(self):
69	n/a	self.assertTrue(issubclass(sqlite.OperationalError, sqlite.DatabaseError),
70	n/a	"OperationalError is not a subclass of DatabaseError")
71	n/a
72	n/a	def CheckIntegrityError(self):
73	n/a	self.assertTrue(issubclass(sqlite.IntegrityError, sqlite.DatabaseError),
74	n/a	"IntegrityError is not a subclass of DatabaseError")
75	n/a
76	n/a	def CheckInternalError(self):
77	n/a	self.assertTrue(issubclass(sqlite.InternalError, sqlite.DatabaseError),
78	n/a	"InternalError is not a subclass of DatabaseError")
79	n/a
80	n/a	def CheckProgrammingError(self):
81	n/a	self.assertTrue(issubclass(sqlite.ProgrammingError, sqlite.DatabaseError),
82	n/a	"ProgrammingError is not a subclass of DatabaseError")
83	n/a
84	n/a	def CheckNotSupportedError(self):
85	n/a	self.assertTrue(issubclass(sqlite.NotSupportedError,
86	n/a	sqlite.DatabaseError),
87	n/a	"NotSupportedError is not a subclass of DatabaseError")
88	n/a
89	n/a	class ConnectionTests(unittest.TestCase):
90	n/a
91	n/a	def setUp(self):
92	n/a	self.cx = sqlite.connect(":memory:")
93	n/a	cu = self.cx.cursor()
94	n/a	cu.execute("create table test(id integer primary key, name text)")
95	n/a	cu.execute("insert into test(name) values (?)", ("foo",))
96	n/a
97	n/a	def tearDown(self):
98	n/a	self.cx.close()
99	n/a
100	n/a	def CheckCommit(self):
101	n/a	self.cx.commit()
102	n/a
103	n/a	def CheckCommitAfterNoChanges(self):
104	n/a	"""
105	n/a	A commit should also work when no changes were made to the database.
106	n/a	"""
107	n/a	self.cx.commit()
108	n/a	self.cx.commit()
109	n/a
110	n/a	def CheckRollback(self):
111	n/a	self.cx.rollback()
112	n/a
113	n/a	def CheckRollbackAfterNoChanges(self):
114	n/a	"""
115	n/a	A rollback should also work when no changes were made to the database.
116	n/a	"""
117	n/a	self.cx.rollback()
118	n/a	self.cx.rollback()
119	n/a
120	n/a	def CheckCursor(self):
121	n/a	cu = self.cx.cursor()
122	n/a
123	n/a	def CheckFailedOpen(self):
124	n/a	YOU_CANNOT_OPEN_THIS = "/foo/bar/bla/23534/mydb.db"
125	n/a	with self.assertRaises(sqlite.OperationalError):
126	n/a	con = sqlite.connect(YOU_CANNOT_OPEN_THIS)
127	n/a
128	n/a	def CheckClose(self):
129	n/a	self.cx.close()
130	n/a
131	n/a	def CheckExceptions(self):
132	n/a	# Optional DB-API extension.
133	n/a	self.assertEqual(self.cx.Warning, sqlite.Warning)
134	n/a	self.assertEqual(self.cx.Error, sqlite.Error)
135	n/a	self.assertEqual(self.cx.InterfaceError, sqlite.InterfaceError)
136	n/a	self.assertEqual(self.cx.DatabaseError, sqlite.DatabaseError)
137	n/a	self.assertEqual(self.cx.DataError, sqlite.DataError)
138	n/a	self.assertEqual(self.cx.OperationalError, sqlite.OperationalError)
139	n/a	self.assertEqual(self.cx.IntegrityError, sqlite.IntegrityError)
140	n/a	self.assertEqual(self.cx.InternalError, sqlite.InternalError)
141	n/a	self.assertEqual(self.cx.ProgrammingError, sqlite.ProgrammingError)
142	n/a	self.assertEqual(self.cx.NotSupportedError, sqlite.NotSupportedError)
143	n/a
144	n/a	def CheckInTransaction(self):
145	n/a	# Can't use db from setUp because we want to test initial state.
146	n/a	cx = sqlite.connect(":memory:")
147	n/a	cu = cx.cursor()
148	n/a	self.assertEqual(cx.in_transaction, False)
149	n/a	cu.execute("create table transactiontest(id integer primary key, name text)")
150	n/a	self.assertEqual(cx.in_transaction, False)
151	n/a	cu.execute("insert into transactiontest(name) values (?)", ("foo",))
152	n/a	self.assertEqual(cx.in_transaction, True)
153	n/a	cu.execute("select name from transactiontest where name=?", ["foo"])
154	n/a	row = cu.fetchone()
155	n/a	self.assertEqual(cx.in_transaction, True)
156	n/a	cx.commit()
157	n/a	self.assertEqual(cx.in_transaction, False)
158	n/a	cu.execute("select name from transactiontest where name=?", ["foo"])
159	n/a	row = cu.fetchone()
160	n/a	self.assertEqual(cx.in_transaction, False)
161	n/a
162	n/a	def CheckInTransactionRO(self):
163	n/a	with self.assertRaises(AttributeError):
164	n/a	self.cx.in_transaction = True
165	n/a
166	n/a	def CheckOpenUri(self):
167	n/a	if sqlite.sqlite_version_info < (3, 7, 7):
168	n/a	with self.assertRaises(sqlite.NotSupportedError):
169	n/a	sqlite.connect(':memory:', uri=True)
170	n/a	return
171	n/a	self.addCleanup(unlink, TESTFN)
172	n/a	with sqlite.connect(TESTFN) as cx:
173	n/a	cx.execute('create table test(id integer)')
174	n/a	with sqlite.connect('file:' + TESTFN, uri=True) as cx:
175	n/a	cx.execute('insert into test(id) values(0)')
176	n/a	with sqlite.connect('file:' + TESTFN + '?mode=ro', uri=True) as cx:
177	n/a	with self.assertRaises(sqlite.OperationalError):
178	n/a	cx.execute('insert into test(id) values(1)')
179	n/a
180	n/a	@unittest.skipIf(sqlite.sqlite_version_info >= (3, 3, 1),
181	n/a	'needs sqlite versions older than 3.3.1')
182	n/a	def CheckSameThreadErrorOnOldVersion(self):
183	n/a	with self.assertRaises(sqlite.NotSupportedError) as cm:
184	n/a	sqlite.connect(':memory:', check_same_thread=False)
185	n/a	self.assertEqual(str(cm.exception), 'shared connections not available')
186	n/a
187	n/a	class CursorTests(unittest.TestCase):
188	n/a	def setUp(self):
189	n/a	self.cx = sqlite.connect(":memory:")
190	n/a	self.cu = self.cx.cursor()
191	n/a	self.cu.execute(
192	n/a	"create table test(id integer primary key, name text, "
193	n/a	"income number, unique_test text unique)"
194	n/a	)
195	n/a	self.cu.execute("insert into test(name) values (?)", ("foo",))
196	n/a
197	n/a	def tearDown(self):
198	n/a	self.cu.close()
199	n/a	self.cx.close()
200	n/a
201	n/a	def CheckExecuteNoArgs(self):
202	n/a	self.cu.execute("delete from test")
203	n/a
204	n/a	def CheckExecuteIllegalSql(self):
205	n/a	with self.assertRaises(sqlite.OperationalError):
206	n/a	self.cu.execute("select asdf")
207	n/a
208	n/a	def CheckExecuteTooMuchSql(self):
209	n/a	with self.assertRaises(sqlite.Warning):
210	n/a	self.cu.execute("select 5+4; select 4+5")
211	n/a
212	n/a	def CheckExecuteTooMuchSql2(self):
213	n/a	self.cu.execute("select 5+4; -- foo bar")
214	n/a
215	n/a	def CheckExecuteTooMuchSql3(self):
216	n/a	self.cu.execute("""
217	n/a	select 5+4;
218	n/a
219	n/a	/*
220	n/a	foo
221	n/a	*/
222	n/a	""")
223	n/a
224	n/a	def CheckExecuteWrongSqlArg(self):
225	n/a	with self.assertRaises(ValueError):
226	n/a	self.cu.execute(42)
227	n/a
228	n/a	def CheckExecuteArgInt(self):
229	n/a	self.cu.execute("insert into test(id) values (?)", (42,))
230	n/a
231	n/a	def CheckExecuteArgFloat(self):
232	n/a	self.cu.execute("insert into test(income) values (?)", (2500.32,))
233	n/a
234	n/a	def CheckExecuteArgString(self):
235	n/a	self.cu.execute("insert into test(name) values (?)", ("Hugo",))
236	n/a
237	n/a	def CheckExecuteArgStringWithZeroByte(self):
238	n/a	self.cu.execute("insert into test(name) values (?)", ("Hu\x00go",))
239	n/a
240	n/a	self.cu.execute("select name from test where id=?", (self.cu.lastrowid,))
241	n/a	row = self.cu.fetchone()
242	n/a	self.assertEqual(row[0], "Hu\x00go")
243	n/a
244	n/a	def CheckExecuteNonIterable(self):
245	n/a	with self.assertRaises(ValueError) as cm:
246	n/a	self.cu.execute("insert into test(id) values (?)", 42)
247	n/a	self.assertEqual(str(cm.exception), 'parameters are of unsupported type')
248	n/a
249	n/a	def CheckExecuteWrongNoOfArgs1(self):
250	n/a	# too many parameters
251	n/a	with self.assertRaises(sqlite.ProgrammingError):
252	n/a	self.cu.execute("insert into test(id) values (?)", (17, "Egon"))
253	n/a
254	n/a	def CheckExecuteWrongNoOfArgs2(self):
255	n/a	# too little parameters
256	n/a	with self.assertRaises(sqlite.ProgrammingError):
257	n/a	self.cu.execute("insert into test(id) values (?)")
258	n/a
259	n/a	def CheckExecuteWrongNoOfArgs3(self):
260	n/a	# no parameters, parameters are needed
261	n/a	with self.assertRaises(sqlite.ProgrammingError):
262	n/a	self.cu.execute("insert into test(id) values (?)")
263	n/a
264	n/a	def CheckExecuteParamList(self):
265	n/a	self.cu.execute("insert into test(name) values ('foo')")
266	n/a	self.cu.execute("select name from test where name=?", ["foo"])
267	n/a	row = self.cu.fetchone()
268	n/a	self.assertEqual(row[0], "foo")
269	n/a
270	n/a	def CheckExecuteParamSequence(self):
271	n/a	class L(object):
272	n/a	def __len__(self):
273	n/a	return 1
274	n/a	def __getitem__(self, x):
275	n/a	assert x == 0
276	n/a	return "foo"
277	n/a
278	n/a	self.cu.execute("insert into test(name) values ('foo')")
279	n/a	self.cu.execute("select name from test where name=?", L())
280	n/a	row = self.cu.fetchone()
281	n/a	self.assertEqual(row[0], "foo")
282	n/a
283	n/a	def CheckExecuteDictMapping(self):
284	n/a	self.cu.execute("insert into test(name) values ('foo')")
285	n/a	self.cu.execute("select name from test where name=:name", {"name": "foo"})
286	n/a	row = self.cu.fetchone()
287	n/a	self.assertEqual(row[0], "foo")
288	n/a
289	n/a	def CheckExecuteDictMapping_Mapping(self):
290	n/a	class D(dict):
291	n/a	def __missing__(self, key):
292	n/a	return "foo"
293	n/a
294	n/a	self.cu.execute("insert into test(name) values ('foo')")
295	n/a	self.cu.execute("select name from test where name=:name", D())
296	n/a	row = self.cu.fetchone()
297	n/a	self.assertEqual(row[0], "foo")
298	n/a
299	n/a	def CheckExecuteDictMappingTooLittleArgs(self):
300	n/a	self.cu.execute("insert into test(name) values ('foo')")
301	n/a	with self.assertRaises(sqlite.ProgrammingError):
302	n/a	self.cu.execute("select name from test where name=:name and id=:id", {"name": "foo"})
303	n/a
304	n/a	def CheckExecuteDictMappingNoArgs(self):
305	n/a	self.cu.execute("insert into test(name) values ('foo')")
306	n/a	with self.assertRaises(sqlite.ProgrammingError):
307	n/a	self.cu.execute("select name from test where name=:name")
308	n/a
309	n/a	def CheckExecuteDictMappingUnnamed(self):
310	n/a	self.cu.execute("insert into test(name) values ('foo')")
311	n/a	with self.assertRaises(sqlite.ProgrammingError):
312	n/a	self.cu.execute("select name from test where name=?", {"name": "foo"})
313	n/a
314	n/a	def CheckClose(self):
315	n/a	self.cu.close()
316	n/a
317	n/a	def CheckRowcountExecute(self):
318	n/a	self.cu.execute("delete from test")
319	n/a	self.cu.execute("insert into test(name) values ('foo')")
320	n/a	self.cu.execute("insert into test(name) values ('foo')")
321	n/a	self.cu.execute("update test set name='bar'")
322	n/a	self.assertEqual(self.cu.rowcount, 2)
323	n/a
324	n/a	def CheckRowcountSelect(self):
325	n/a	"""
326	n/a	pysqlite does not know the rowcount of SELECT statements, because we
327	n/a	don't fetch all rows after executing the select statement. The rowcount
328	n/a	has thus to be -1.
329	n/a	"""
330	n/a	self.cu.execute("select 5 union select 6")
331	n/a	self.assertEqual(self.cu.rowcount, -1)
332	n/a
333	n/a	def CheckRowcountExecutemany(self):
334	n/a	self.cu.execute("delete from test")
335	n/a	self.cu.executemany("insert into test(name) values (?)", [(1,), (2,), (3,)])
336	n/a	self.assertEqual(self.cu.rowcount, 3)
337	n/a
338	n/a	def CheckTotalChanges(self):
339	n/a	self.cu.execute("insert into test(name) values ('foo')")
340	n/a	self.cu.execute("insert into test(name) values ('foo')")
341	n/a	self.assertLess(2, self.cx.total_changes, msg='total changes reported wrong value')
342	n/a
343	n/a	# Checks for executemany:
344	n/a	# Sequences are required by the DB-API, iterators
345	n/a	# enhancements in pysqlite.
346	n/a
347	n/a	def CheckExecuteManySequence(self):
348	n/a	self.cu.executemany("insert into test(income) values (?)", [(x,) for x in range(100, 110)])
349	n/a
350	n/a	def CheckExecuteManyIterator(self):
351	n/a	class MyIter:
352	n/a	def __init__(self):
353	n/a	self.value = 5
354	n/a
355	n/a	def __next__(self):
356	n/a	if self.value == 10:
357	n/a	raise StopIteration
358	n/a	else:
359	n/a	self.value += 1
360	n/a	return (self.value,)
361	n/a
362	n/a	self.cu.executemany("insert into test(income) values (?)", MyIter())
363	n/a
364	n/a	def CheckExecuteManyGenerator(self):
365	n/a	def mygen():
366	n/a	for i in range(5):
367	n/a	yield (i,)
368	n/a
369	n/a	self.cu.executemany("insert into test(income) values (?)", mygen())
370	n/a
371	n/a	def CheckExecuteManyWrongSqlArg(self):
372	n/a	with self.assertRaises(ValueError):
373	n/a	self.cu.executemany(42, [(3,)])
374	n/a
375	n/a	def CheckExecuteManySelect(self):
376	n/a	with self.assertRaises(sqlite.ProgrammingError):
377	n/a	self.cu.executemany("select ?", [(3,)])
378	n/a
379	n/a	def CheckExecuteManyNotIterable(self):
380	n/a	with self.assertRaises(TypeError):
381	n/a	self.cu.executemany("insert into test(income) values (?)", 42)
382	n/a
383	n/a	def CheckFetchIter(self):
384	n/a	# Optional DB-API extension.
385	n/a	self.cu.execute("delete from test")
386	n/a	self.cu.execute("insert into test(id) values (?)", (5,))
387	n/a	self.cu.execute("insert into test(id) values (?)", (6,))
388	n/a	self.cu.execute("select id from test order by id")
389	n/a	lst = []
390	n/a	for row in self.cu:
391	n/a	lst.append(row[0])
392	n/a	self.assertEqual(lst[0], 5)
393	n/a	self.assertEqual(lst[1], 6)
394	n/a
395	n/a	def CheckFetchone(self):
396	n/a	self.cu.execute("select name from test")
397	n/a	row = self.cu.fetchone()
398	n/a	self.assertEqual(row[0], "foo")
399	n/a	row = self.cu.fetchone()
400	n/a	self.assertEqual(row, None)
401	n/a
402	n/a	def CheckFetchoneNoStatement(self):
403	n/a	cur = self.cx.cursor()
404	n/a	row = cur.fetchone()
405	n/a	self.assertEqual(row, None)
406	n/a
407	n/a	def CheckArraySize(self):
408	n/a	# must default ot 1
409	n/a	self.assertEqual(self.cu.arraysize, 1)
410	n/a
411	n/a	# now set to 2
412	n/a	self.cu.arraysize = 2
413	n/a
414	n/a	# now make the query return 3 rows
415	n/a	self.cu.execute("delete from test")
416	n/a	self.cu.execute("insert into test(name) values ('A')")
417	n/a	self.cu.execute("insert into test(name) values ('B')")
418	n/a	self.cu.execute("insert into test(name) values ('C')")
419	n/a	self.cu.execute("select name from test")
420	n/a	res = self.cu.fetchmany()
421	n/a
422	n/a	self.assertEqual(len(res), 2)
423	n/a
424	n/a	def CheckFetchmany(self):
425	n/a	self.cu.execute("select name from test")
426	n/a	res = self.cu.fetchmany(100)
427	n/a	self.assertEqual(len(res), 1)
428	n/a	res = self.cu.fetchmany(100)
429	n/a	self.assertEqual(res, [])
430	n/a
431	n/a	def CheckFetchmanyKwArg(self):
432	n/a	"""Checks if fetchmany works with keyword arguments"""
433	n/a	self.cu.execute("select name from test")
434	n/a	res = self.cu.fetchmany(size=100)
435	n/a	self.assertEqual(len(res), 1)
436	n/a
437	n/a	def CheckFetchall(self):
438	n/a	self.cu.execute("select name from test")
439	n/a	res = self.cu.fetchall()
440	n/a	self.assertEqual(len(res), 1)
441	n/a	res = self.cu.fetchall()
442	n/a	self.assertEqual(res, [])
443	n/a
444	n/a	def CheckSetinputsizes(self):
445	n/a	self.cu.setinputsizes([3, 4, 5])
446	n/a
447	n/a	def CheckSetoutputsize(self):
448	n/a	self.cu.setoutputsize(5, 0)
449	n/a
450	n/a	def CheckSetoutputsizeNoColumn(self):
451	n/a	self.cu.setoutputsize(42)
452	n/a
453	n/a	def CheckCursorConnection(self):
454	n/a	# Optional DB-API extension.
455	n/a	self.assertEqual(self.cu.connection, self.cx)
456	n/a
457	n/a	def CheckWrongCursorCallable(self):
458	n/a	with self.assertRaises(TypeError):
459	n/a	def f(): pass
460	n/a	cur = self.cx.cursor(f)
461	n/a
462	n/a	def CheckCursorWrongClass(self):
463	n/a	class Foo: pass
464	n/a	foo = Foo()
465	n/a	with self.assertRaises(TypeError):
466	n/a	cur = sqlite.Cursor(foo)
467	n/a
468	n/a	def CheckLastRowIDOnReplace(self):
469	n/a	"""
470	n/a	INSERT OR REPLACE and REPLACE INTO should produce the same behavior.
471	n/a	"""
472	n/a	sql = '{} INTO test(id, unique_test) VALUES (?, ?)'
473	n/a	for statement in ('INSERT OR REPLACE', 'REPLACE'):
474	n/a	with self.subTest(statement=statement):
475	n/a	self.cu.execute(sql.format(statement), (1, 'foo'))
476	n/a	self.assertEqual(self.cu.lastrowid, 1)
477	n/a
478	n/a	def CheckLastRowIDOnIgnore(self):
479	n/a	self.cu.execute(
480	n/a	"insert or ignore into test(unique_test) values (?)",
481	n/a	('test',))
482	n/a	self.assertEqual(self.cu.lastrowid, 2)
483	n/a	self.cu.execute(
484	n/a	"insert or ignore into test(unique_test) values (?)",
485	n/a	('test',))
486	n/a	self.assertEqual(self.cu.lastrowid, 2)
487	n/a
488	n/a	def CheckLastRowIDInsertOR(self):
489	n/a	results = []
490	n/a	for statement in ('FAIL', 'ABORT', 'ROLLBACK'):
491	n/a	sql = 'INSERT OR {} INTO test(unique_test) VALUES (?)'
492	n/a	with self.subTest(statement='INSERT OR {}'.format(statement)):
493	n/a	self.cu.execute(sql.format(statement), (statement,))
494	n/a	results.append((statement, self.cu.lastrowid))
495	n/a	with self.assertRaises(sqlite.IntegrityError):
496	n/a	self.cu.execute(sql.format(statement), (statement,))
497	n/a	results.append((statement, self.cu.lastrowid))
498	n/a	expected = [
499	n/a	('FAIL', 2), ('FAIL', 2),
500	n/a	('ABORT', 3), ('ABORT', 3),
501	n/a	('ROLLBACK', 4), ('ROLLBACK', 4),
502	n/a	]
503	n/a	self.assertEqual(results, expected)
504	n/a
505	n/a
506	n/a	@unittest.skipUnless(threading, 'This test requires threading.')
507	n/a	class ThreadTests(unittest.TestCase):
508	n/a	def setUp(self):
509	n/a	self.con = sqlite.connect(":memory:")
510	n/a	self.cur = self.con.cursor()
511	n/a	self.cur.execute("create table test(id integer primary key, name text, bin binary, ratio number, ts timestamp)")
512	n/a
513	n/a	def tearDown(self):
514	n/a	self.cur.close()
515	n/a	self.con.close()
516	n/a
517	n/a	def CheckConCursor(self):
518	n/a	def run(con, errors):
519	n/a	try:
520	n/a	cur = con.cursor()
521	n/a	errors.append("did not raise ProgrammingError")
522	n/a	return
523	n/a	except sqlite.ProgrammingError:
524	n/a	return
525	n/a	except:
526	n/a	errors.append("raised wrong exception")
527	n/a
528	n/a	errors = []
529	n/a	t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors})
530	n/a	t.start()
531	n/a	t.join()
532	n/a	if len(errors) > 0:
533	n/a	self.fail("\n".join(errors))
534	n/a
535	n/a	def CheckConCommit(self):
536	n/a	def run(con, errors):
537	n/a	try:
538	n/a	con.commit()
539	n/a	errors.append("did not raise ProgrammingError")
540	n/a	return
541	n/a	except sqlite.ProgrammingError:
542	n/a	return
543	n/a	except:
544	n/a	errors.append("raised wrong exception")
545	n/a
546	n/a	errors = []
547	n/a	t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors})
548	n/a	t.start()
549	n/a	t.join()
550	n/a	if len(errors) > 0:
551	n/a	self.fail("\n".join(errors))
552	n/a
553	n/a	def CheckConRollback(self):
554	n/a	def run(con, errors):
555	n/a	try:
556	n/a	con.rollback()
557	n/a	errors.append("did not raise ProgrammingError")
558	n/a	return
559	n/a	except sqlite.ProgrammingError:
560	n/a	return
561	n/a	except:
562	n/a	errors.append("raised wrong exception")
563	n/a
564	n/a	errors = []
565	n/a	t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors})
566	n/a	t.start()
567	n/a	t.join()
568	n/a	if len(errors) > 0:
569	n/a	self.fail("\n".join(errors))
570	n/a
571	n/a	def CheckConClose(self):
572	n/a	def run(con, errors):
573	n/a	try:
574	n/a	con.close()
575	n/a	errors.append("did not raise ProgrammingError")
576	n/a	return
577	n/a	except sqlite.ProgrammingError:
578	n/a	return
579	n/a	except:
580	n/a	errors.append("raised wrong exception")
581	n/a
582	n/a	errors = []
583	n/a	t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors})
584	n/a	t.start()
585	n/a	t.join()
586	n/a	if len(errors) > 0:
587	n/a	self.fail("\n".join(errors))
588	n/a
589	n/a	def CheckCurImplicitBegin(self):
590	n/a	def run(cur, errors):
591	n/a	try:
592	n/a	cur.execute("insert into test(name) values ('a')")
593	n/a	errors.append("did not raise ProgrammingError")
594	n/a	return
595	n/a	except sqlite.ProgrammingError:
596	n/a	return
597	n/a	except:
598	n/a	errors.append("raised wrong exception")
599	n/a
600	n/a	errors = []
601	n/a	t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors})
602	n/a	t.start()
603	n/a	t.join()
604	n/a	if len(errors) > 0:
605	n/a	self.fail("\n".join(errors))
606	n/a
607	n/a	def CheckCurClose(self):
608	n/a	def run(cur, errors):
609	n/a	try:
610	n/a	cur.close()
611	n/a	errors.append("did not raise ProgrammingError")
612	n/a	return
613	n/a	except sqlite.ProgrammingError:
614	n/a	return
615	n/a	except:
616	n/a	errors.append("raised wrong exception")
617	n/a
618	n/a	errors = []
619	n/a	t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors})
620	n/a	t.start()
621	n/a	t.join()
622	n/a	if len(errors) > 0:
623	n/a	self.fail("\n".join(errors))
624	n/a
625	n/a	def CheckCurExecute(self):
626	n/a	def run(cur, errors):
627	n/a	try:
628	n/a	cur.execute("select name from test")
629	n/a	errors.append("did not raise ProgrammingError")
630	n/a	return
631	n/a	except sqlite.ProgrammingError:
632	n/a	return
633	n/a	except:
634	n/a	errors.append("raised wrong exception")
635	n/a
636	n/a	errors = []
637	n/a	self.cur.execute("insert into test(name) values ('a')")
638	n/a	t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors})
639	n/a	t.start()
640	n/a	t.join()
641	n/a	if len(errors) > 0:
642	n/a	self.fail("\n".join(errors))
643	n/a
644	n/a	def CheckCurIterNext(self):
645	n/a	def run(cur, errors):
646	n/a	try:
647	n/a	row = cur.fetchone()
648	n/a	errors.append("did not raise ProgrammingError")
649	n/a	return
650	n/a	except sqlite.ProgrammingError:
651	n/a	return
652	n/a	except:
653	n/a	errors.append("raised wrong exception")
654	n/a
655	n/a	errors = []
656	n/a	self.cur.execute("insert into test(name) values ('a')")
657	n/a	self.cur.execute("select name from test")
658	n/a	t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors})
659	n/a	t.start()
660	n/a	t.join()
661	n/a	if len(errors) > 0:
662	n/a	self.fail("\n".join(errors))
663	n/a
664	n/a	class ConstructorTests(unittest.TestCase):
665	n/a	def CheckDate(self):
666	n/a	d = sqlite.Date(2004, 10, 28)
667	n/a
668	n/a	def CheckTime(self):
669	n/a	t = sqlite.Time(12, 39, 35)
670	n/a
671	n/a	def CheckTimestamp(self):
672	n/a	ts = sqlite.Timestamp(2004, 10, 28, 12, 39, 35)
673	n/a
674	n/a	def CheckDateFromTicks(self):
675	n/a	d = sqlite.DateFromTicks(42)
676	n/a
677	n/a	def CheckTimeFromTicks(self):
678	n/a	t = sqlite.TimeFromTicks(42)
679	n/a
680	n/a	def CheckTimestampFromTicks(self):
681	n/a	ts = sqlite.TimestampFromTicks(42)
682	n/a
683	n/a	def CheckBinary(self):
684	n/a	b = sqlite.Binary(b"\0'")
685	n/a
686	n/a	class ExtensionTests(unittest.TestCase):
687	n/a	def CheckScriptStringSql(self):
688	n/a	con = sqlite.connect(":memory:")
689	n/a	cur = con.cursor()
690	n/a	cur.executescript("""
691	n/a	-- bla bla
692	n/a	/* a stupid comment */
693	n/a	create table a(i);
694	n/a	insert into a(i) values (5);
695	n/a	""")
696	n/a	cur.execute("select i from a")
697	n/a	res = cur.fetchone()[0]
698	n/a	self.assertEqual(res, 5)
699	n/a
700	n/a	def CheckScriptSyntaxError(self):
701	n/a	con = sqlite.connect(":memory:")
702	n/a	cur = con.cursor()
703	n/a	with self.assertRaises(sqlite.OperationalError):
704	n/a	cur.executescript("create table test(x); asdf; create table test2(x)")
705	n/a
706	n/a	def CheckScriptErrorNormal(self):
707	n/a	con = sqlite.connect(":memory:")
708	n/a	cur = con.cursor()
709	n/a	with self.assertRaises(sqlite.OperationalError):
710	n/a	cur.executescript("create table test(sadfsadfdsa); select foo from hurz;")
711	n/a
712	n/a	def CheckCursorExecutescriptAsBytes(self):
713	n/a	con = sqlite.connect(":memory:")
714	n/a	cur = con.cursor()
715	n/a	with self.assertRaises(ValueError) as cm:
716	n/a	cur.executescript(b"create table test(foo); insert into test(foo) values (5);")
717	n/a	self.assertEqual(str(cm.exception), 'script argument must be unicode.')
718	n/a
719	n/a	def CheckConnectionExecute(self):
720	n/a	con = sqlite.connect(":memory:")
721	n/a	result = con.execute("select 5").fetchone()[0]
722	n/a	self.assertEqual(result, 5, "Basic test of Connection.execute")
723	n/a
724	n/a	def CheckConnectionExecutemany(self):
725	n/a	con = sqlite.connect(":memory:")
726	n/a	con.execute("create table test(foo)")
727	n/a	con.executemany("insert into test(foo) values (?)", [(3,), (4,)])
728	n/a	result = con.execute("select foo from test order by foo").fetchall()
729	n/a	self.assertEqual(result[0][0], 3, "Basic test of Connection.executemany")
730	n/a	self.assertEqual(result[1][0], 4, "Basic test of Connection.executemany")
731	n/a
732	n/a	def CheckConnectionExecutescript(self):
733	n/a	con = sqlite.connect(":memory:")
734	n/a	con.executescript("create table test(foo); insert into test(foo) values (5);")
735	n/a	result = con.execute("select foo from test").fetchone()[0]
736	n/a	self.assertEqual(result, 5, "Basic test of Connection.executescript")
737	n/a
738	n/a	class ClosedConTests(unittest.TestCase):
739	n/a	def CheckClosedConCursor(self):
740	n/a	con = sqlite.connect(":memory:")
741	n/a	con.close()
742	n/a	with self.assertRaises(sqlite.ProgrammingError):
743	n/a	cur = con.cursor()
744	n/a
745	n/a	def CheckClosedConCommit(self):
746	n/a	con = sqlite.connect(":memory:")
747	n/a	con.close()
748	n/a	with self.assertRaises(sqlite.ProgrammingError):
749	n/a	con.commit()
750	n/a
751	n/a	def CheckClosedConRollback(self):
752	n/a	con = sqlite.connect(":memory:")
753	n/a	con.close()
754	n/a	with self.assertRaises(sqlite.ProgrammingError):
755	n/a	con.rollback()
756	n/a
757	n/a	def CheckClosedCurExecute(self):
758	n/a	con = sqlite.connect(":memory:")
759	n/a	cur = con.cursor()
760	n/a	con.close()
761	n/a	with self.assertRaises(sqlite.ProgrammingError):
762	n/a	cur.execute("select 4")
763	n/a
764	n/a	def CheckClosedCreateFunction(self):
765	n/a	con = sqlite.connect(":memory:")
766	n/a	con.close()
767	n/a	def f(x): return 17
768	n/a	with self.assertRaises(sqlite.ProgrammingError):
769	n/a	con.create_function("foo", 1, f)
770	n/a
771	n/a	def CheckClosedCreateAggregate(self):
772	n/a	con = sqlite.connect(":memory:")
773	n/a	con.close()
774	n/a	class Agg:
775	n/a	def __init__(self):
776	n/a	pass
777	n/a	def step(self, x):
778	n/a	pass
779	n/a	def finalize(self):
780	n/a	return 17
781	n/a	with self.assertRaises(sqlite.ProgrammingError):
782	n/a	con.create_aggregate("foo", 1, Agg)
783	n/a
784	n/a	def CheckClosedSetAuthorizer(self):
785	n/a	con = sqlite.connect(":memory:")
786	n/a	con.close()
787	n/a	def authorizer(*args):
788	n/a	return sqlite.DENY
789	n/a	with self.assertRaises(sqlite.ProgrammingError):
790	n/a	con.set_authorizer(authorizer)
791	n/a
792	n/a	def CheckClosedSetProgressCallback(self):
793	n/a	con = sqlite.connect(":memory:")
794	n/a	con.close()
795	n/a	def progress(): pass
796	n/a	with self.assertRaises(sqlite.ProgrammingError):
797	n/a	con.set_progress_handler(progress, 100)
798	n/a
799	n/a	def CheckClosedCall(self):
800	n/a	con = sqlite.connect(":memory:")
801	n/a	con.close()
802	n/a	with self.assertRaises(sqlite.ProgrammingError):
803	n/a	con()
804	n/a
805	n/a	class ClosedCurTests(unittest.TestCase):
806	n/a	def CheckClosed(self):
807	n/a	con = sqlite.connect(":memory:")
808	n/a	cur = con.cursor()
809	n/a	cur.close()
810	n/a
811	n/a	for method_name in ("execute", "executemany", "executescript", "fetchall", "fetchmany", "fetchone"):
812	n/a	if method_name in ("execute", "executescript"):
813	n/a	params = ("select 4 union select 5",)
814	n/a	elif method_name == "executemany":
815	n/a	params = ("insert into foo(bar) values (?)", [(3,), (4,)])
816	n/a	else:
817	n/a	params = []
818	n/a
819	n/a	with self.assertRaises(sqlite.ProgrammingError):
820	n/a	method = getattr(cur, method_name)
821	n/a	method(*params)
822	n/a
823	n/a
824	n/a	class SqliteOnConflictTests(unittest.TestCase):
825	n/a	"""
826	n/a	Tests for SQLite's "insert on conflict" feature.
827	n/a
828	n/a	See https://www.sqlite.org/lang_conflict.html for details.
829	n/a	"""
830	n/a
831	n/a	def setUp(self):
832	n/a	self.cx = sqlite.connect(":memory:")
833	n/a	self.cu = self.cx.cursor()
834	n/a	self.cu.execute("""
835	n/a	CREATE TABLE test(
836	n/a	id INTEGER PRIMARY KEY, name TEXT, unique_name TEXT UNIQUE
837	n/a	);
838	n/a	""")
839	n/a
840	n/a	def tearDown(self):
841	n/a	self.cu.close()
842	n/a	self.cx.close()
843	n/a
844	n/a	def CheckOnConflictRollbackWithExplicitTransaction(self):
845	n/a	self.cx.isolation_level = None # autocommit mode
846	n/a	self.cu = self.cx.cursor()
847	n/a	# Start an explicit transaction.
848	n/a	self.cu.execute("BEGIN")
849	n/a	self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')")
850	n/a	self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')")
851	n/a	with self.assertRaises(sqlite.IntegrityError):
852	n/a	self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')")
853	n/a	# Use connection to commit.
854	n/a	self.cx.commit()
855	n/a	self.cu.execute("SELECT name, unique_name from test")
856	n/a	# Transaction should have rolled back and nothing should be in table.
857	n/a	self.assertEqual(self.cu.fetchall(), [])
858	n/a
859	n/a	def CheckOnConflictAbortRaisesWithExplicitTransactions(self):
860	n/a	# Abort cancels the current sql statement but doesn't change anything
861	n/a	# about the current transaction.
862	n/a	self.cx.isolation_level = None # autocommit mode
863	n/a	self.cu = self.cx.cursor()
864	n/a	# Start an explicit transaction.
865	n/a	self.cu.execute("BEGIN")
866	n/a	self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')")
867	n/a	self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')")
868	n/a	with self.assertRaises(sqlite.IntegrityError):
869	n/a	self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')")
870	n/a	self.cx.commit()
871	n/a	self.cu.execute("SELECT name, unique_name FROM test")
872	n/a	# Expect the first two inserts to work, third to do nothing.
873	n/a	self.assertEqual(self.cu.fetchall(), [('abort_test', None), (None, 'foo',)])
874	n/a
875	n/a	def CheckOnConflictRollbackWithoutTransaction(self):
876	n/a	# Start of implicit transaction
877	n/a	self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')")
878	n/a	self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')")
879	n/a	with self.assertRaises(sqlite.IntegrityError):
880	n/a	self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')")
881	n/a	self.cu.execute("SELECT name, unique_name FROM test")
882	n/a	# Implicit transaction is rolled back on error.
883	n/a	self.assertEqual(self.cu.fetchall(), [])
884	n/a
885	n/a	def CheckOnConflictAbortRaisesWithoutTransactions(self):
886	n/a	# Abort cancels the current sql statement but doesn't change anything
887	n/a	# about the current transaction.
888	n/a	self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')")
889	n/a	self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')")
890	n/a	with self.assertRaises(sqlite.IntegrityError):
891	n/a	self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')")
892	n/a	# Make sure all other values were inserted.
893	n/a	self.cu.execute("SELECT name, unique_name FROM test")
894	n/a	self.assertEqual(self.cu.fetchall(), [('abort_test', None), (None, 'foo',)])
895	n/a
896	n/a	def CheckOnConflictFail(self):
897	n/a	self.cu.execute("INSERT OR FAIL INTO test(unique_name) VALUES ('foo')")
898	n/a	with self.assertRaises(sqlite.IntegrityError):
899	n/a	self.cu.execute("INSERT OR FAIL INTO test(unique_name) VALUES ('foo')")
900	n/a	self.assertEqual(self.cu.fetchall(), [])
901	n/a
902	n/a	def CheckOnConflictIgnore(self):
903	n/a	self.cu.execute("INSERT OR IGNORE INTO test(unique_name) VALUES ('foo')")
904	n/a	# Nothing should happen.
905	n/a	self.cu.execute("INSERT OR IGNORE INTO test(unique_name) VALUES ('foo')")
906	n/a	self.cu.execute("SELECT unique_name FROM test")
907	n/a	self.assertEqual(self.cu.fetchall(), [('foo',)])
908	n/a
909	n/a	def CheckOnConflictReplace(self):
910	n/a	self.cu.execute("INSERT OR REPLACE INTO test(name, unique_name) VALUES ('Data!', 'foo')")
911	n/a	# There shouldn't be an IntegrityError exception.
912	n/a	self.cu.execute("INSERT OR REPLACE INTO test(name, unique_name) VALUES ('Very different data!', 'foo')")
913	n/a	self.cu.execute("SELECT name, unique_name FROM test")
914	n/a	self.assertEqual(self.cu.fetchall(), [('Very different data!', 'foo')])
915	n/a
916	n/a
917	n/a	def suite():
918	n/a	module_suite = unittest.makeSuite(ModuleTests, "Check")
919	n/a	connection_suite = unittest.makeSuite(ConnectionTests, "Check")
920	n/a	cursor_suite = unittest.makeSuite(CursorTests, "Check")
921	n/a	thread_suite = unittest.makeSuite(ThreadTests, "Check")
922	n/a	constructor_suite = unittest.makeSuite(ConstructorTests, "Check")
923	n/a	ext_suite = unittest.makeSuite(ExtensionTests, "Check")
924	n/a	closed_con_suite = unittest.makeSuite(ClosedConTests, "Check")
925	n/a	closed_cur_suite = unittest.makeSuite(ClosedCurTests, "Check")
926	n/a	on_conflict_suite = unittest.makeSuite(SqliteOnConflictTests, "Check")
927	n/a	return unittest.TestSuite((
928	n/a	module_suite, connection_suite, cursor_suite, thread_suite,
929	n/a	constructor_suite, ext_suite, closed_con_suite, closed_cur_suite,
930	n/a	on_conflict_suite,
931	n/a	))
932	n/a
933	n/a	def test():
934	n/a	runner = unittest.TextTestRunner()
935	n/a	runner.run(suite())
936	n/a
937	n/a	if __name__ == "__main__":
938	n/a	test()