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

#	count	content
1	n/a	#-- coding: iso-8859-1 --
2	n/a	# pysqlite2/test/factory.py: tests for the various factories in pysqlite
3	n/a	#
4	n/a	# Copyright (C) 2005-2007 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	from collections.abc import Sequence
27	n/a
28	n/a	class MyConnection(sqlite.Connection):
29	n/a	def __init__(self, args, *kwargs):
30	n/a	sqlite.Connection.__init__(self, args, *kwargs)
31	n/a
32	n/a	def dict_factory(cursor, row):
33	n/a	d = {}
34	n/a	for idx, col in enumerate(cursor.description):
35	n/a	d[col[0]] = row[idx]
36	n/a	return d
37	n/a
38	n/a	class MyCursor(sqlite.Cursor):
39	n/a	def __init__(self, args, *kwargs):
40	n/a	sqlite.Cursor.__init__(self, args, *kwargs)
41	n/a	self.row_factory = dict_factory
42	n/a
43	n/a	class ConnectionFactoryTests(unittest.TestCase):
44	n/a	def setUp(self):
45	n/a	self.con = sqlite.connect(":memory:", factory=MyConnection)
46	n/a
47	n/a	def tearDown(self):
48	n/a	self.con.close()
49	n/a
50	n/a	def CheckIsInstance(self):
51	n/a	self.assertIsInstance(self.con, MyConnection)
52	n/a
53	n/a	class CursorFactoryTests(unittest.TestCase):
54	n/a	def setUp(self):
55	n/a	self.con = sqlite.connect(":memory:")
56	n/a
57	n/a	def tearDown(self):
58	n/a	self.con.close()
59	n/a
60	n/a	def CheckIsInstance(self):
61	n/a	cur = self.con.cursor()
62	n/a	self.assertIsInstance(cur, sqlite.Cursor)
63	n/a	cur = self.con.cursor(MyCursor)
64	n/a	self.assertIsInstance(cur, MyCursor)
65	n/a	cur = self.con.cursor(factory=lambda con: MyCursor(con))
66	n/a	self.assertIsInstance(cur, MyCursor)
67	n/a
68	n/a	def CheckInvalidFactory(self):
69	n/a	# not a callable at all
70	n/a	self.assertRaises(TypeError, self.con.cursor, None)
71	n/a	# invalid callable with not exact one argument
72	n/a	self.assertRaises(TypeError, self.con.cursor, lambda: None)
73	n/a	# invalid callable returning non-cursor
74	n/a	self.assertRaises(TypeError, self.con.cursor, lambda con: None)
75	n/a
76	n/a	class RowFactoryTestsBackwardsCompat(unittest.TestCase):
77	n/a	def setUp(self):
78	n/a	self.con = sqlite.connect(":memory:")
79	n/a
80	n/a	def CheckIsProducedByFactory(self):
81	n/a	cur = self.con.cursor(factory=MyCursor)
82	n/a	cur.execute("select 4+5 as foo")
83	n/a	row = cur.fetchone()
84	n/a	self.assertIsInstance(row, dict)
85	n/a	cur.close()
86	n/a
87	n/a	def tearDown(self):
88	n/a	self.con.close()
89	n/a
90	n/a	class RowFactoryTests(unittest.TestCase):
91	n/a	def setUp(self):
92	n/a	self.con = sqlite.connect(":memory:")
93	n/a
94	n/a	def CheckCustomFactory(self):
95	n/a	self.con.row_factory = lambda cur, row: list(row)
96	n/a	row = self.con.execute("select 1, 2").fetchone()
97	n/a	self.assertIsInstance(row, list)
98	n/a
99	n/a	def CheckSqliteRowIndex(self):
100	n/a	self.con.row_factory = sqlite.Row
101	n/a	row = self.con.execute("select 1 as a, 2 as b").fetchone()
102	n/a	self.assertIsInstance(row, sqlite.Row)
103	n/a
104	n/a	col1, col2 = row["a"], row["b"]
105	n/a	self.assertEqual(col1, 1, "by name: wrong result for column 'a'")
106	n/a	self.assertEqual(col2, 2, "by name: wrong result for column 'a'")
107	n/a
108	n/a	col1, col2 = row["A"], row["B"]
109	n/a	self.assertEqual(col1, 1, "by name: wrong result for column 'A'")
110	n/a	self.assertEqual(col2, 2, "by name: wrong result for column 'B'")
111	n/a
112	n/a	self.assertEqual(row[0], 1, "by index: wrong result for column 0")
113	n/a	self.assertEqual(row[1], 2, "by index: wrong result for column 1")
114	n/a	self.assertEqual(row[-1], 2, "by index: wrong result for column -1")
115	n/a	self.assertEqual(row[-2], 1, "by index: wrong result for column -2")
116	n/a
117	n/a	with self.assertRaises(IndexError):
118	n/a	row['c']
119	n/a	with self.assertRaises(IndexError):
120	n/a	row[2]
121	n/a	with self.assertRaises(IndexError):
122	n/a	row[-3]
123	n/a	with self.assertRaises(IndexError):
124	n/a	row[2**1000]
125	n/a
126	n/a	def CheckSqliteRowSlice(self):
127	n/a	# A sqlite.Row can be sliced like a list.
128	n/a	self.con.row_factory = sqlite.Row
129	n/a	row = self.con.execute("select 1, 2, 3, 4").fetchone()
130	n/a	self.assertEqual(row[0:0], ())
131	n/a	self.assertEqual(row[0:1], (1,))
132	n/a	self.assertEqual(row[1:3], (2, 3))
133	n/a	self.assertEqual(row[3:1], ())
134	n/a	# Explicit bounds are optional.
135	n/a	self.assertEqual(row[1:], (2, 3, 4))
136	n/a	self.assertEqual(row[:3], (1, 2, 3))
137	n/a	# Slices can use negative indices.
138	n/a	self.assertEqual(row[-2:-1], (3,))
139	n/a	self.assertEqual(row[-2:], (3, 4))
140	n/a	# Slicing supports steps.
141	n/a	self.assertEqual(row[0:4:2], (1, 3))
142	n/a	self.assertEqual(row[3:0:-2], (4, 2))
143	n/a
144	n/a	def CheckSqliteRowIter(self):
145	n/a	"""Checks if the row object is iterable"""
146	n/a	self.con.row_factory = sqlite.Row
147	n/a	row = self.con.execute("select 1 as a, 2 as b").fetchone()
148	n/a	for col in row:
149	n/a	pass
150	n/a
151	n/a	def CheckSqliteRowAsTuple(self):
152	n/a	"""Checks if the row object can be converted to a tuple"""
153	n/a	self.con.row_factory = sqlite.Row
154	n/a	row = self.con.execute("select 1 as a, 2 as b").fetchone()
155	n/a	t = tuple(row)
156	n/a	self.assertEqual(t, (row['a'], row['b']))
157	n/a
158	n/a	def CheckSqliteRowAsDict(self):
159	n/a	"""Checks if the row object can be correctly converted to a dictionary"""
160	n/a	self.con.row_factory = sqlite.Row
161	n/a	row = self.con.execute("select 1 as a, 2 as b").fetchone()
162	n/a	d = dict(row)
163	n/a	self.assertEqual(d["a"], row["a"])
164	n/a	self.assertEqual(d["b"], row["b"])
165	n/a
166	n/a	def CheckSqliteRowHashCmp(self):
167	n/a	"""Checks if the row object compares and hashes correctly"""
168	n/a	self.con.row_factory = sqlite.Row
169	n/a	row_1 = self.con.execute("select 1 as a, 2 as b").fetchone()
170	n/a	row_2 = self.con.execute("select 1 as a, 2 as b").fetchone()
171	n/a	row_3 = self.con.execute("select 1 as a, 3 as b").fetchone()
172	n/a
173	n/a	self.assertEqual(row_1, row_1)
174	n/a	self.assertEqual(row_1, row_2)
175	n/a	self.assertTrue(row_2 != row_3)
176	n/a
177	n/a	self.assertFalse(row_1 != row_1)
178	n/a	self.assertFalse(row_1 != row_2)
179	n/a	self.assertFalse(row_2 == row_3)
180	n/a
181	n/a	self.assertEqual(row_1, row_2)
182	n/a	self.assertEqual(hash(row_1), hash(row_2))
183	n/a	self.assertNotEqual(row_1, row_3)
184	n/a	self.assertNotEqual(hash(row_1), hash(row_3))
185	n/a
186	n/a	def CheckSqliteRowAsSequence(self):
187	n/a	""" Checks if the row object can act like a sequence """
188	n/a	self.con.row_factory = sqlite.Row
189	n/a	row = self.con.execute("select 1 as a, 2 as b").fetchone()
190	n/a
191	n/a	as_tuple = tuple(row)
192	n/a	self.assertEqual(list(reversed(row)), list(reversed(as_tuple)))
193	n/a	self.assertIsInstance(row, Sequence)
194	n/a
195	n/a	def CheckFakeCursorClass(self):
196	n/a	# Issue #24257: Incorrect use of PyObject_IsInstance() caused
197	n/a	# segmentation fault.
198	n/a	# Issue #27861: Also applies for cursor factory.
199	n/a	class FakeCursor(str):
200	n/a	__class__ = sqlite.Cursor
201	n/a	self.con.row_factory = sqlite.Row
202	n/a	self.assertRaises(TypeError, self.con.cursor, FakeCursor)
203	n/a	self.assertRaises(TypeError, sqlite.Row, FakeCursor(), ())
204	n/a
205	n/a	def tearDown(self):
206	n/a	self.con.close()
207	n/a
208	n/a	class TextFactoryTests(unittest.TestCase):
209	n/a	def setUp(self):
210	n/a	self.con = sqlite.connect(":memory:")
211	n/a
212	n/a	def CheckUnicode(self):
213	n/a	austria = "Österreich"
214	n/a	row = self.con.execute("select ?", (austria,)).fetchone()
215	n/a	self.assertEqual(type(row[0]), str, "type of row[0] must be unicode")
216	n/a
217	n/a	def CheckString(self):
218	n/a	self.con.text_factory = bytes
219	n/a	austria = "Österreich"
220	n/a	row = self.con.execute("select ?", (austria,)).fetchone()
221	n/a	self.assertEqual(type(row[0]), bytes, "type of row[0] must be bytes")
222	n/a	self.assertEqual(row[0], austria.encode("utf-8"), "column must equal original data in UTF-8")
223	n/a
224	n/a	def CheckCustom(self):
225	n/a	self.con.text_factory = lambda x: str(x, "utf-8", "ignore")
226	n/a	austria = "Österreich"
227	n/a	row = self.con.execute("select ?", (austria,)).fetchone()
228	n/a	self.assertEqual(type(row[0]), str, "type of row[0] must be unicode")
229	n/a	self.assertTrue(row[0].endswith("reich"), "column must contain original data")
230	n/a
231	n/a	def CheckOptimizedUnicode(self):
232	n/a	# In py3k, str objects are always returned when text_factory
233	n/a	# is OptimizedUnicode
234	n/a	self.con.text_factory = sqlite.OptimizedUnicode
235	n/a	austria = "Österreich"
236	n/a	germany = "Deutchland"
237	n/a	a_row = self.con.execute("select ?", (austria,)).fetchone()
238	n/a	d_row = self.con.execute("select ?", (germany,)).fetchone()
239	n/a	self.assertEqual(type(a_row[0]), str, "type of non-ASCII row must be str")
240	n/a	self.assertEqual(type(d_row[0]), str, "type of ASCII-only row must be str")
241	n/a
242	n/a	def tearDown(self):
243	n/a	self.con.close()
244	n/a
245	n/a	class TextFactoryTestsWithEmbeddedZeroBytes(unittest.TestCase):
246	n/a	def setUp(self):
247	n/a	self.con = sqlite.connect(":memory:")
248	n/a	self.con.execute("create table test (value text)")
249	n/a	self.con.execute("insert into test (value) values (?)", ("a\x00b",))
250	n/a
251	n/a	def CheckString(self):
252	n/a	# text_factory defaults to str
253	n/a	row = self.con.execute("select value from test").fetchone()
254	n/a	self.assertIs(type(row[0]), str)
255	n/a	self.assertEqual(row[0], "a\x00b")
256	n/a
257	n/a	def CheckBytes(self):
258	n/a	self.con.text_factory = bytes
259	n/a	row = self.con.execute("select value from test").fetchone()
260	n/a	self.assertIs(type(row[0]), bytes)
261	n/a	self.assertEqual(row[0], b"a\x00b")
262	n/a
263	n/a	def CheckBytearray(self):
264	n/a	self.con.text_factory = bytearray
265	n/a	row = self.con.execute("select value from test").fetchone()
266	n/a	self.assertIs(type(row[0]), bytearray)
267	n/a	self.assertEqual(row[0], b"a\x00b")
268	n/a
269	n/a	def CheckCustom(self):
270	n/a	# A custom factory should receive a bytes argument
271	n/a	self.con.text_factory = lambda x: x
272	n/a	row = self.con.execute("select value from test").fetchone()
273	n/a	self.assertIs(type(row[0]), bytes)
274	n/a	self.assertEqual(row[0], b"a\x00b")
275	n/a
276	n/a	def tearDown(self):
277	n/a	self.con.close()
278	n/a
279	n/a	def suite():
280	n/a	connection_suite = unittest.makeSuite(ConnectionFactoryTests, "Check")
281	n/a	cursor_suite = unittest.makeSuite(CursorFactoryTests, "Check")
282	n/a	row_suite_compat = unittest.makeSuite(RowFactoryTestsBackwardsCompat, "Check")
283	n/a	row_suite = unittest.makeSuite(RowFactoryTests, "Check")
284	n/a	text_suite = unittest.makeSuite(TextFactoryTests, "Check")
285	n/a	text_zero_bytes_suite = unittest.makeSuite(TextFactoryTestsWithEmbeddedZeroBytes, "Check")
286	n/a	return unittest.TestSuite((connection_suite, cursor_suite, row_suite_compat, row_suite, text_suite, text_zero_bytes_suite))
287	n/a
288	n/a	def test():
289	n/a	runner = unittest.TextTestRunner()
290	n/a	runner.run(suite())
291	n/a
292	n/a	if __name__ == "__main__":
293	n/a	test()