»Core Development>Code coverage>Lib/sqlite3/test/dbapi.py

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

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