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

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

#countcontent
1n/a#-*- coding: iso-8859-1 -*-
2n/a# pysqlite2/test/types.py: tests for type conversion and detection
3n/a#
4n/a# Copyright (C) 2005 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 datetime
25n/aimport unittest
26n/aimport sqlite3 as sqlite
27n/atry:
28n/a import zlib
29n/aexcept ImportError:
30n/a zlib = None
31n/a
32n/a
33n/aclass SqliteTypeTests(unittest.TestCase):
34n/a def setUp(self):
35n/a self.con = sqlite.connect(":memory:")
36n/a self.cur = self.con.cursor()
37n/a self.cur.execute("create table test(i integer, s varchar, f number, b blob)")
38n/a
39n/a def tearDown(self):
40n/a self.cur.close()
41n/a self.con.close()
42n/a
43n/a def CheckString(self):
44n/a self.cur.execute("insert into test(s) values (?)", ("Österreich",))
45n/a self.cur.execute("select s from test")
46n/a row = self.cur.fetchone()
47n/a self.assertEqual(row[0], "Österreich")
48n/a
49n/a def CheckSmallInt(self):
50n/a self.cur.execute("insert into test(i) values (?)", (42,))
51n/a self.cur.execute("select i from test")
52n/a row = self.cur.fetchone()
53n/a self.assertEqual(row[0], 42)
54n/a
55n/a def CheckLargeInt(self):
56n/a num = 2**40
57n/a self.cur.execute("insert into test(i) values (?)", (num,))
58n/a self.cur.execute("select i from test")
59n/a row = self.cur.fetchone()
60n/a self.assertEqual(row[0], num)
61n/a
62n/a def CheckFloat(self):
63n/a val = 3.14
64n/a self.cur.execute("insert into test(f) values (?)", (val,))
65n/a self.cur.execute("select f from test")
66n/a row = self.cur.fetchone()
67n/a self.assertEqual(row[0], val)
68n/a
69n/a def CheckBlob(self):
70n/a sample = b"Guglhupf"
71n/a val = memoryview(sample)
72n/a self.cur.execute("insert into test(b) values (?)", (val,))
73n/a self.cur.execute("select b from test")
74n/a row = self.cur.fetchone()
75n/a self.assertEqual(row[0], sample)
76n/a
77n/a def CheckUnicodeExecute(self):
78n/a self.cur.execute("select 'Österreich'")
79n/a row = self.cur.fetchone()
80n/a self.assertEqual(row[0], "Österreich")
81n/a
82n/aclass DeclTypesTests(unittest.TestCase):
83n/a class Foo:
84n/a def __init__(self, _val):
85n/a if isinstance(_val, bytes):
86n/a # sqlite3 always calls __init__ with a bytes created from a
87n/a # UTF-8 string when __conform__ was used to store the object.
88n/a _val = _val.decode('utf-8')
89n/a self.val = _val
90n/a
91n/a def __eq__(self, other):
92n/a if not isinstance(other, DeclTypesTests.Foo):
93n/a return NotImplemented
94n/a return self.val == other.val
95n/a
96n/a def __conform__(self, protocol):
97n/a if protocol is sqlite.PrepareProtocol:
98n/a return self.val
99n/a else:
100n/a return None
101n/a
102n/a def __str__(self):
103n/a return "<%s>" % self.val
104n/a
105n/a def setUp(self):
106n/a self.con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_DECLTYPES)
107n/a self.cur = self.con.cursor()
108n/a self.cur.execute("create table test(i int, s str, f float, b bool, u unicode, foo foo, bin blob, n1 number, n2 number(5))")
109n/a
110n/a # override float, make them always return the same number
111n/a sqlite.converters["FLOAT"] = lambda x: 47.2
112n/a
113n/a # and implement two custom ones
114n/a sqlite.converters["BOOL"] = lambda x: bool(int(x))
115n/a sqlite.converters["FOO"] = DeclTypesTests.Foo
116n/a sqlite.converters["WRONG"] = lambda x: "WRONG"
117n/a sqlite.converters["NUMBER"] = float
118n/a
119n/a def tearDown(self):
120n/a del sqlite.converters["FLOAT"]
121n/a del sqlite.converters["BOOL"]
122n/a del sqlite.converters["FOO"]
123n/a del sqlite.converters["NUMBER"]
124n/a self.cur.close()
125n/a self.con.close()
126n/a
127n/a def CheckString(self):
128n/a # default
129n/a self.cur.execute("insert into test(s) values (?)", ("foo",))
130n/a self.cur.execute('select s as "s [WRONG]" from test')
131n/a row = self.cur.fetchone()
132n/a self.assertEqual(row[0], "foo")
133n/a
134n/a def CheckSmallInt(self):
135n/a # default
136n/a self.cur.execute("insert into test(i) values (?)", (42,))
137n/a self.cur.execute("select i from test")
138n/a row = self.cur.fetchone()
139n/a self.assertEqual(row[0], 42)
140n/a
141n/a def CheckLargeInt(self):
142n/a # default
143n/a num = 2**40
144n/a self.cur.execute("insert into test(i) values (?)", (num,))
145n/a self.cur.execute("select i from test")
146n/a row = self.cur.fetchone()
147n/a self.assertEqual(row[0], num)
148n/a
149n/a def CheckFloat(self):
150n/a # custom
151n/a val = 3.14
152n/a self.cur.execute("insert into test(f) values (?)", (val,))
153n/a self.cur.execute("select f from test")
154n/a row = self.cur.fetchone()
155n/a self.assertEqual(row[0], 47.2)
156n/a
157n/a def CheckBool(self):
158n/a # custom
159n/a self.cur.execute("insert into test(b) values (?)", (False,))
160n/a self.cur.execute("select b from test")
161n/a row = self.cur.fetchone()
162n/a self.assertEqual(row[0], False)
163n/a
164n/a self.cur.execute("delete from test")
165n/a self.cur.execute("insert into test(b) values (?)", (True,))
166n/a self.cur.execute("select b from test")
167n/a row = self.cur.fetchone()
168n/a self.assertEqual(row[0], True)
169n/a
170n/a def CheckUnicode(self):
171n/a # default
172n/a val = "\xd6sterreich"
173n/a self.cur.execute("insert into test(u) values (?)", (val,))
174n/a self.cur.execute("select u from test")
175n/a row = self.cur.fetchone()
176n/a self.assertEqual(row[0], val)
177n/a
178n/a def CheckFoo(self):
179n/a val = DeclTypesTests.Foo("bla")
180n/a self.cur.execute("insert into test(foo) values (?)", (val,))
181n/a self.cur.execute("select foo from test")
182n/a row = self.cur.fetchone()
183n/a self.assertEqual(row[0], val)
184n/a
185n/a def CheckUnsupportedSeq(self):
186n/a class Bar: pass
187n/a val = Bar()
188n/a with self.assertRaises(sqlite.InterfaceError):
189n/a self.cur.execute("insert into test(f) values (?)", (val,))
190n/a
191n/a def CheckUnsupportedDict(self):
192n/a class Bar: pass
193n/a val = Bar()
194n/a with self.assertRaises(sqlite.InterfaceError):
195n/a self.cur.execute("insert into test(f) values (:val)", {"val": val})
196n/a
197n/a def CheckBlob(self):
198n/a # default
199n/a sample = b"Guglhupf"
200n/a val = memoryview(sample)
201n/a self.cur.execute("insert into test(bin) values (?)", (val,))
202n/a self.cur.execute("select bin from test")
203n/a row = self.cur.fetchone()
204n/a self.assertEqual(row[0], sample)
205n/a
206n/a def CheckNumber1(self):
207n/a self.cur.execute("insert into test(n1) values (5)")
208n/a value = self.cur.execute("select n1 from test").fetchone()[0]
209n/a # if the converter is not used, it's an int instead of a float
210n/a self.assertEqual(type(value), float)
211n/a
212n/a def CheckNumber2(self):
213n/a """Checks whether converter names are cut off at '(' characters"""
214n/a self.cur.execute("insert into test(n2) values (5)")
215n/a value = self.cur.execute("select n2 from test").fetchone()[0]
216n/a # if the converter is not used, it's an int instead of a float
217n/a self.assertEqual(type(value), float)
218n/a
219n/aclass ColNamesTests(unittest.TestCase):
220n/a def setUp(self):
221n/a self.con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_COLNAMES)
222n/a self.cur = self.con.cursor()
223n/a self.cur.execute("create table test(x foo)")
224n/a
225n/a sqlite.converters["FOO"] = lambda x: "[%s]" % x.decode("ascii")
226n/a sqlite.converters["BAR"] = lambda x: "<%s>" % x.decode("ascii")
227n/a sqlite.converters["EXC"] = lambda x: 5/0
228n/a sqlite.converters["B1B1"] = lambda x: "MARKER"
229n/a
230n/a def tearDown(self):
231n/a del sqlite.converters["FOO"]
232n/a del sqlite.converters["BAR"]
233n/a del sqlite.converters["EXC"]
234n/a del sqlite.converters["B1B1"]
235n/a self.cur.close()
236n/a self.con.close()
237n/a
238n/a def CheckDeclTypeNotUsed(self):
239n/a """
240n/a Assures that the declared type is not used when PARSE_DECLTYPES
241n/a is not set.
242n/a """
243n/a self.cur.execute("insert into test(x) values (?)", ("xxx",))
244n/a self.cur.execute("select x from test")
245n/a val = self.cur.fetchone()[0]
246n/a self.assertEqual(val, "xxx")
247n/a
248n/a def CheckNone(self):
249n/a self.cur.execute("insert into test(x) values (?)", (None,))
250n/a self.cur.execute("select x from test")
251n/a val = self.cur.fetchone()[0]
252n/a self.assertEqual(val, None)
253n/a
254n/a def CheckColName(self):
255n/a self.cur.execute("insert into test(x) values (?)", ("xxx",))
256n/a self.cur.execute('select x as "x [bar]" from test')
257n/a val = self.cur.fetchone()[0]
258n/a self.assertEqual(val, "<xxx>")
259n/a
260n/a # Check if the stripping of colnames works. Everything after the first
261n/a # whitespace should be stripped.
262n/a self.assertEqual(self.cur.description[0][0], "x")
263n/a
264n/a def CheckCaseInConverterName(self):
265n/a self.cur.execute("select 'other' as \"x [b1b1]\"")
266n/a val = self.cur.fetchone()[0]
267n/a self.assertEqual(val, "MARKER")
268n/a
269n/a def CheckCursorDescriptionNoRow(self):
270n/a """
271n/a cursor.description should at least provide the column name(s), even if
272n/a no row returned.
273n/a """
274n/a self.cur.execute("select * from test where 0 = 1")
275n/a self.assertEqual(self.cur.description[0][0], "x")
276n/a
277n/a def CheckCursorDescriptionInsert(self):
278n/a self.cur.execute("insert into test values (1)")
279n/a self.assertIsNone(self.cur.description)
280n/a
281n/a
282n/a@unittest.skipIf(sqlite.sqlite_version_info < (3, 8, 3), "CTEs not supported")
283n/aclass CommonTableExpressionTests(unittest.TestCase):
284n/a
285n/a def setUp(self):
286n/a self.con = sqlite.connect(":memory:")
287n/a self.cur = self.con.cursor()
288n/a self.cur.execute("create table test(x foo)")
289n/a
290n/a def tearDown(self):
291n/a self.cur.close()
292n/a self.con.close()
293n/a
294n/a def CheckCursorDescriptionCTESimple(self):
295n/a self.cur.execute("with one as (select 1) select * from one")
296n/a self.assertIsNotNone(self.cur.description)
297n/a self.assertEqual(self.cur.description[0][0], "1")
298n/a
299n/a def CheckCursorDescriptionCTESMultipleColumns(self):
300n/a self.cur.execute("insert into test values(1)")
301n/a self.cur.execute("insert into test values(2)")
302n/a self.cur.execute("with testCTE as (select * from test) select * from testCTE")
303n/a self.assertIsNotNone(self.cur.description)
304n/a self.assertEqual(self.cur.description[0][0], "x")
305n/a
306n/a def CheckCursorDescriptionCTE(self):
307n/a self.cur.execute("insert into test values (1)")
308n/a self.cur.execute("with bar as (select * from test) select * from test where x = 1")
309n/a self.assertIsNotNone(self.cur.description)
310n/a self.assertEqual(self.cur.description[0][0], "x")
311n/a self.cur.execute("with bar as (select * from test) select * from test where x = 2")
312n/a self.assertIsNotNone(self.cur.description)
313n/a self.assertEqual(self.cur.description[0][0], "x")
314n/a
315n/a
316n/aclass ObjectAdaptationTests(unittest.TestCase):
317n/a def cast(obj):
318n/a return float(obj)
319n/a cast = staticmethod(cast)
320n/a
321n/a def setUp(self):
322n/a self.con = sqlite.connect(":memory:")
323n/a try:
324n/a del sqlite.adapters[int]
325n/a except:
326n/a pass
327n/a sqlite.register_adapter(int, ObjectAdaptationTests.cast)
328n/a self.cur = self.con.cursor()
329n/a
330n/a def tearDown(self):
331n/a del sqlite.adapters[(int, sqlite.PrepareProtocol)]
332n/a self.cur.close()
333n/a self.con.close()
334n/a
335n/a def CheckCasterIsUsed(self):
336n/a self.cur.execute("select ?", (4,))
337n/a val = self.cur.fetchone()[0]
338n/a self.assertEqual(type(val), float)
339n/a
340n/a@unittest.skipUnless(zlib, "requires zlib")
341n/aclass BinaryConverterTests(unittest.TestCase):
342n/a def convert(s):
343n/a return zlib.decompress(s)
344n/a convert = staticmethod(convert)
345n/a
346n/a def setUp(self):
347n/a self.con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_COLNAMES)
348n/a sqlite.register_converter("bin", BinaryConverterTests.convert)
349n/a
350n/a def tearDown(self):
351n/a self.con.close()
352n/a
353n/a def CheckBinaryInputForConverter(self):
354n/a testdata = b"abcdefg" * 10
355n/a result = self.con.execute('select ? as "x [bin]"', (memoryview(zlib.compress(testdata)),)).fetchone()[0]
356n/a self.assertEqual(testdata, result)
357n/a
358n/aclass DateTimeTests(unittest.TestCase):
359n/a def setUp(self):
360n/a self.con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_DECLTYPES)
361n/a self.cur = self.con.cursor()
362n/a self.cur.execute("create table test(d date, ts timestamp)")
363n/a
364n/a def tearDown(self):
365n/a self.cur.close()
366n/a self.con.close()
367n/a
368n/a def CheckSqliteDate(self):
369n/a d = sqlite.Date(2004, 2, 14)
370n/a self.cur.execute("insert into test(d) values (?)", (d,))
371n/a self.cur.execute("select d from test")
372n/a d2 = self.cur.fetchone()[0]
373n/a self.assertEqual(d, d2)
374n/a
375n/a def CheckSqliteTimestamp(self):
376n/a ts = sqlite.Timestamp(2004, 2, 14, 7, 15, 0)
377n/a self.cur.execute("insert into test(ts) values (?)", (ts,))
378n/a self.cur.execute("select ts from test")
379n/a ts2 = self.cur.fetchone()[0]
380n/a self.assertEqual(ts, ts2)
381n/a
382n/a @unittest.skipIf(sqlite.sqlite_version_info < (3, 1),
383n/a 'the date functions are available on 3.1 or later')
384n/a def CheckSqlTimestamp(self):
385n/a now = datetime.datetime.utcnow()
386n/a self.cur.execute("insert into test(ts) values (current_timestamp)")
387n/a self.cur.execute("select ts from test")
388n/a ts = self.cur.fetchone()[0]
389n/a self.assertEqual(type(ts), datetime.datetime)
390n/a self.assertEqual(ts.year, now.year)
391n/a
392n/a def CheckDateTimeSubSeconds(self):
393n/a ts = sqlite.Timestamp(2004, 2, 14, 7, 15, 0, 500000)
394n/a self.cur.execute("insert into test(ts) values (?)", (ts,))
395n/a self.cur.execute("select ts from test")
396n/a ts2 = self.cur.fetchone()[0]
397n/a self.assertEqual(ts, ts2)
398n/a
399n/a def CheckDateTimeSubSecondsFloatingPoint(self):
400n/a ts = sqlite.Timestamp(2004, 2, 14, 7, 15, 0, 510241)
401n/a self.cur.execute("insert into test(ts) values (?)", (ts,))
402n/a self.cur.execute("select ts from test")
403n/a ts2 = self.cur.fetchone()[0]
404n/a self.assertEqual(ts, ts2)
405n/a
406n/adef suite():
407n/a sqlite_type_suite = unittest.makeSuite(SqliteTypeTests, "Check")
408n/a decltypes_type_suite = unittest.makeSuite(DeclTypesTests, "Check")
409n/a colnames_type_suite = unittest.makeSuite(ColNamesTests, "Check")
410n/a adaptation_suite = unittest.makeSuite(ObjectAdaptationTests, "Check")
411n/a bin_suite = unittest.makeSuite(BinaryConverterTests, "Check")
412n/a date_suite = unittest.makeSuite(DateTimeTests, "Check")
413n/a cte_suite = unittest.makeSuite(CommonTableExpressionTests, "Check")
414n/a return unittest.TestSuite((sqlite_type_suite, decltypes_type_suite, colnames_type_suite, adaptation_suite, bin_suite, date_suite, cte_suite))
415n/a
416n/adef test():
417n/a runner = unittest.TextTestRunner()
418n/a runner.run(suite())
419n/a
420n/aif __name__ == "__main__":
421n/a test()