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

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