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

#	count	content
1	n/a	#-- coding: iso-8859-1 --
2	n/a	# pysqlite2/test/userfunctions.py: tests for user-defined functions and
3	n/a	# aggregates.
4	n/a	#
5	n/a	# Copyright (C) 2005-2007 Gerhard Häring <gh@ghaering.de>
6	n/a	#
7	n/a	# This file is part of pysqlite.
8	n/a	#
9	n/a	# This software is provided 'as-is', without any express or implied
10	n/a	# warranty. In no event will the authors be held liable for any damages
11	n/a	# arising from the use of this software.
12	n/a	#
13	n/a	# Permission is granted to anyone to use this software for any purpose,
14	n/a	# including commercial applications, and to alter it and redistribute it
15	n/a	# freely, subject to the following restrictions:
16	n/a	#
17	n/a	# 1. The origin of this software must not be misrepresented; you must not
18	n/a	# claim that you wrote the original software. If you use this software
19	n/a	# in a product, an acknowledgment in the product documentation would be
20	n/a	# appreciated but is not required.
21	n/a	# 2. Altered source versions must be plainly marked as such, and must not be
22	n/a	# misrepresented as being the original software.
23	n/a	# 3. This notice may not be removed or altered from any source distribution.
24	n/a
25	n/a	import unittest
26	n/a	import sqlite3 as sqlite
27	n/a
28	n/a	def func_returntext():
29	n/a	return "foo"
30	n/a	def func_returnunicode():
31	n/a	return "bar"
32	n/a	def func_returnint():
33	n/a	return 42
34	n/a	def func_returnfloat():
35	n/a	return 3.14
36	n/a	def func_returnnull():
37	n/a	return None
38	n/a	def func_returnblob():
39	n/a	return b"blob"
40	n/a	def func_returnlonglong():
41	n/a	return 1<<31
42	n/a	def func_raiseexception():
43	n/a	5/0
44	n/a
45	n/a	def func_isstring(v):
46	n/a	return type(v) is str
47	n/a	def func_isint(v):
48	n/a	return type(v) is int
49	n/a	def func_isfloat(v):
50	n/a	return type(v) is float
51	n/a	def func_isnone(v):
52	n/a	return type(v) is type(None)
53	n/a	def func_isblob(v):
54	n/a	return isinstance(v, (bytes, memoryview))
55	n/a	def func_islonglong(v):
56	n/a	return isinstance(v, int) and v >= 1<<31
57	n/a
58	n/a	def func(*args):
59	n/a	return len(args)
60	n/a
61	n/a	class AggrNoStep:
62	n/a	def __init__(self):
63	n/a	pass
64	n/a
65	n/a	def finalize(self):
66	n/a	return 1
67	n/a
68	n/a	class AggrNoFinalize:
69	n/a	def __init__(self):
70	n/a	pass
71	n/a
72	n/a	def step(self, x):
73	n/a	pass
74	n/a
75	n/a	class AggrExceptionInInit:
76	n/a	def __init__(self):
77	n/a	5/0
78	n/a
79	n/a	def step(self, x):
80	n/a	pass
81	n/a
82	n/a	def finalize(self):
83	n/a	pass
84	n/a
85	n/a	class AggrExceptionInStep:
86	n/a	def __init__(self):
87	n/a	pass
88	n/a
89	n/a	def step(self, x):
90	n/a	5/0
91	n/a
92	n/a	def finalize(self):
93	n/a	return 42
94	n/a
95	n/a	class AggrExceptionInFinalize:
96	n/a	def __init__(self):
97	n/a	pass
98	n/a
99	n/a	def step(self, x):
100	n/a	pass
101	n/a
102	n/a	def finalize(self):
103	n/a	5/0
104	n/a
105	n/a	class AggrCheckType:
106	n/a	def __init__(self):
107	n/a	self.val = None
108	n/a
109	n/a	def step(self, whichType, val):
110	n/a	theType = {"str": str, "int": int, "float": float, "None": type(None),
111	n/a	"blob": bytes}
112	n/a	self.val = int(theType[whichType] is type(val))
113	n/a
114	n/a	def finalize(self):
115	n/a	return self.val
116	n/a
117	n/a	class AggrCheckTypes:
118	n/a	def __init__(self):
119	n/a	self.val = 0
120	n/a
121	n/a	def step(self, whichType, *vals):
122	n/a	theType = {"str": str, "int": int, "float": float, "None": type(None),
123	n/a	"blob": bytes}
124	n/a	for val in vals:
125	n/a	self.val += int(theType[whichType] is type(val))
126	n/a
127	n/a	def finalize(self):
128	n/a	return self.val
129	n/a
130	n/a	class AggrSum:
131	n/a	def __init__(self):
132	n/a	self.val = 0.0
133	n/a
134	n/a	def step(self, val):
135	n/a	self.val += val
136	n/a
137	n/a	def finalize(self):
138	n/a	return self.val
139	n/a
140	n/a	class FunctionTests(unittest.TestCase):
141	n/a	def setUp(self):
142	n/a	self.con = sqlite.connect(":memory:")
143	n/a
144	n/a	self.con.create_function("returntext", 0, func_returntext)
145	n/a	self.con.create_function("returnunicode", 0, func_returnunicode)
146	n/a	self.con.create_function("returnint", 0, func_returnint)
147	n/a	self.con.create_function("returnfloat", 0, func_returnfloat)
148	n/a	self.con.create_function("returnnull", 0, func_returnnull)
149	n/a	self.con.create_function("returnblob", 0, func_returnblob)
150	n/a	self.con.create_function("returnlonglong", 0, func_returnlonglong)
151	n/a	self.con.create_function("raiseexception", 0, func_raiseexception)
152	n/a
153	n/a	self.con.create_function("isstring", 1, func_isstring)
154	n/a	self.con.create_function("isint", 1, func_isint)
155	n/a	self.con.create_function("isfloat", 1, func_isfloat)
156	n/a	self.con.create_function("isnone", 1, func_isnone)
157	n/a	self.con.create_function("isblob", 1, func_isblob)
158	n/a	self.con.create_function("islonglong", 1, func_islonglong)
159	n/a	self.con.create_function("spam", -1, func)
160	n/a
161	n/a	def tearDown(self):
162	n/a	self.con.close()
163	n/a
164	n/a	def CheckFuncErrorOnCreate(self):
165	n/a	with self.assertRaises(sqlite.OperationalError):
166	n/a	self.con.create_function("bla", -100, lambda x: 2*x)
167	n/a
168	n/a	def CheckFuncRefCount(self):
169	n/a	def getfunc():
170	n/a	def f():
171	n/a	return 1
172	n/a	return f
173	n/a	f = getfunc()
174	n/a	globals()["foo"] = f
175	n/a	# self.con.create_function("reftest", 0, getfunc())
176	n/a	self.con.create_function("reftest", 0, f)
177	n/a	cur = self.con.cursor()
178	n/a	cur.execute("select reftest()")
179	n/a
180	n/a	def CheckFuncReturnText(self):
181	n/a	cur = self.con.cursor()
182	n/a	cur.execute("select returntext()")
183	n/a	val = cur.fetchone()[0]
184	n/a	self.assertEqual(type(val), str)
185	n/a	self.assertEqual(val, "foo")
186	n/a
187	n/a	def CheckFuncReturnUnicode(self):
188	n/a	cur = self.con.cursor()
189	n/a	cur.execute("select returnunicode()")
190	n/a	val = cur.fetchone()[0]
191	n/a	self.assertEqual(type(val), str)
192	n/a	self.assertEqual(val, "bar")
193	n/a
194	n/a	def CheckFuncReturnInt(self):
195	n/a	cur = self.con.cursor()
196	n/a	cur.execute("select returnint()")
197	n/a	val = cur.fetchone()[0]
198	n/a	self.assertEqual(type(val), int)
199	n/a	self.assertEqual(val, 42)
200	n/a
201	n/a	def CheckFuncReturnFloat(self):
202	n/a	cur = self.con.cursor()
203	n/a	cur.execute("select returnfloat()")
204	n/a	val = cur.fetchone()[0]
205	n/a	self.assertEqual(type(val), float)
206	n/a	if val < 3.139 or val > 3.141:
207	n/a	self.fail("wrong value")
208	n/a
209	n/a	def CheckFuncReturnNull(self):
210	n/a	cur = self.con.cursor()
211	n/a	cur.execute("select returnnull()")
212	n/a	val = cur.fetchone()[0]
213	n/a	self.assertEqual(type(val), type(None))
214	n/a	self.assertEqual(val, None)
215	n/a
216	n/a	def CheckFuncReturnBlob(self):
217	n/a	cur = self.con.cursor()
218	n/a	cur.execute("select returnblob()")
219	n/a	val = cur.fetchone()[0]
220	n/a	self.assertEqual(type(val), bytes)
221	n/a	self.assertEqual(val, b"blob")
222	n/a
223	n/a	def CheckFuncReturnLongLong(self):
224	n/a	cur = self.con.cursor()
225	n/a	cur.execute("select returnlonglong()")
226	n/a	val = cur.fetchone()[0]
227	n/a	self.assertEqual(val, 1<<31)
228	n/a
229	n/a	def CheckFuncException(self):
230	n/a	cur = self.con.cursor()
231	n/a	with self.assertRaises(sqlite.OperationalError) as cm:
232	n/a	cur.execute("select raiseexception()")
233	n/a	cur.fetchone()
234	n/a	self.assertEqual(str(cm.exception), 'user-defined function raised exception')
235	n/a
236	n/a	def CheckParamString(self):
237	n/a	cur = self.con.cursor()
238	n/a	cur.execute("select isstring(?)", ("foo",))
239	n/a	val = cur.fetchone()[0]
240	n/a	self.assertEqual(val, 1)
241	n/a
242	n/a	def CheckParamInt(self):
243	n/a	cur = self.con.cursor()
244	n/a	cur.execute("select isint(?)", (42,))
245	n/a	val = cur.fetchone()[0]
246	n/a	self.assertEqual(val, 1)
247	n/a
248	n/a	def CheckParamFloat(self):
249	n/a	cur = self.con.cursor()
250	n/a	cur.execute("select isfloat(?)", (3.14,))
251	n/a	val = cur.fetchone()[0]
252	n/a	self.assertEqual(val, 1)
253	n/a
254	n/a	def CheckParamNone(self):
255	n/a	cur = self.con.cursor()
256	n/a	cur.execute("select isnone(?)", (None,))
257	n/a	val = cur.fetchone()[0]
258	n/a	self.assertEqual(val, 1)
259	n/a
260	n/a	def CheckParamBlob(self):
261	n/a	cur = self.con.cursor()
262	n/a	cur.execute("select isblob(?)", (memoryview(b"blob"),))
263	n/a	val = cur.fetchone()[0]
264	n/a	self.assertEqual(val, 1)
265	n/a
266	n/a	def CheckParamLongLong(self):
267	n/a	cur = self.con.cursor()
268	n/a	cur.execute("select islonglong(?)", (1<<42,))
269	n/a	val = cur.fetchone()[0]
270	n/a	self.assertEqual(val, 1)
271	n/a
272	n/a	def CheckAnyArguments(self):
273	n/a	cur = self.con.cursor()
274	n/a	cur.execute("select spam(?, ?)", (1, 2))
275	n/a	val = cur.fetchone()[0]
276	n/a	self.assertEqual(val, 2)
277	n/a
278	n/a
279	n/a	class AggregateTests(unittest.TestCase):
280	n/a	def setUp(self):
281	n/a	self.con = sqlite.connect(":memory:")
282	n/a	cur = self.con.cursor()
283	n/a	cur.execute("""
284	n/a	create table test(
285	n/a	t text,
286	n/a	i integer,
287	n/a	f float,
288	n/a	n,
289	n/a	b blob
290	n/a	)
291	n/a	""")
292	n/a	cur.execute("insert into test(t, i, f, n, b) values (?, ?, ?, ?, ?)",
293	n/a	("foo", 5, 3.14, None, memoryview(b"blob"),))
294	n/a
295	n/a	self.con.create_aggregate("nostep", 1, AggrNoStep)
296	n/a	self.con.create_aggregate("nofinalize", 1, AggrNoFinalize)
297	n/a	self.con.create_aggregate("excInit", 1, AggrExceptionInInit)
298	n/a	self.con.create_aggregate("excStep", 1, AggrExceptionInStep)
299	n/a	self.con.create_aggregate("excFinalize", 1, AggrExceptionInFinalize)
300	n/a	self.con.create_aggregate("checkType", 2, AggrCheckType)
301	n/a	self.con.create_aggregate("checkTypes", -1, AggrCheckTypes)
302	n/a	self.con.create_aggregate("mysum", 1, AggrSum)
303	n/a
304	n/a	def tearDown(self):
305	n/a	#self.cur.close()
306	n/a	#self.con.close()
307	n/a	pass
308	n/a
309	n/a	def CheckAggrErrorOnCreate(self):
310	n/a	with self.assertRaises(sqlite.OperationalError):
311	n/a	self.con.create_function("bla", -100, AggrSum)
312	n/a
313	n/a	def CheckAggrNoStep(self):
314	n/a	cur = self.con.cursor()
315	n/a	with self.assertRaises(AttributeError) as cm:
316	n/a	cur.execute("select nostep(t) from test")
317	n/a	self.assertEqual(str(cm.exception), "'AggrNoStep' object has no attribute 'step'")
318	n/a
319	n/a	def CheckAggrNoFinalize(self):
320	n/a	cur = self.con.cursor()
321	n/a	with self.assertRaises(sqlite.OperationalError) as cm:
322	n/a	cur.execute("select nofinalize(t) from test")
323	n/a	val = cur.fetchone()[0]
324	n/a	self.assertEqual(str(cm.exception), "user-defined aggregate's 'finalize' method raised error")
325	n/a
326	n/a	def CheckAggrExceptionInInit(self):
327	n/a	cur = self.con.cursor()
328	n/a	with self.assertRaises(sqlite.OperationalError) as cm:
329	n/a	cur.execute("select excInit(t) from test")
330	n/a	val = cur.fetchone()[0]
331	n/a	self.assertEqual(str(cm.exception), "user-defined aggregate's '__init__' method raised error")
332	n/a
333	n/a	def CheckAggrExceptionInStep(self):
334	n/a	cur = self.con.cursor()
335	n/a	with self.assertRaises(sqlite.OperationalError) as cm:
336	n/a	cur.execute("select excStep(t) from test")
337	n/a	val = cur.fetchone()[0]
338	n/a	self.assertEqual(str(cm.exception), "user-defined aggregate's 'step' method raised error")
339	n/a
340	n/a	def CheckAggrExceptionInFinalize(self):
341	n/a	cur = self.con.cursor()
342	n/a	with self.assertRaises(sqlite.OperationalError) as cm:
343	n/a	cur.execute("select excFinalize(t) from test")
344	n/a	val = cur.fetchone()[0]
345	n/a	self.assertEqual(str(cm.exception), "user-defined aggregate's 'finalize' method raised error")
346	n/a
347	n/a	def CheckAggrCheckParamStr(self):
348	n/a	cur = self.con.cursor()
349	n/a	cur.execute("select checkType('str', ?)", ("foo",))
350	n/a	val = cur.fetchone()[0]
351	n/a	self.assertEqual(val, 1)
352	n/a
353	n/a	def CheckAggrCheckParamInt(self):
354	n/a	cur = self.con.cursor()
355	n/a	cur.execute("select checkType('int', ?)", (42,))
356	n/a	val = cur.fetchone()[0]
357	n/a	self.assertEqual(val, 1)
358	n/a
359	n/a	def CheckAggrCheckParamsInt(self):
360	n/a	cur = self.con.cursor()
361	n/a	cur.execute("select checkTypes('int', ?, ?)", (42, 24))
362	n/a	val = cur.fetchone()[0]
363	n/a	self.assertEqual(val, 2)
364	n/a
365	n/a	def CheckAggrCheckParamFloat(self):
366	n/a	cur = self.con.cursor()
367	n/a	cur.execute("select checkType('float', ?)", (3.14,))
368	n/a	val = cur.fetchone()[0]
369	n/a	self.assertEqual(val, 1)
370	n/a
371	n/a	def CheckAggrCheckParamNone(self):
372	n/a	cur = self.con.cursor()
373	n/a	cur.execute("select checkType('None', ?)", (None,))
374	n/a	val = cur.fetchone()[0]
375	n/a	self.assertEqual(val, 1)
376	n/a
377	n/a	def CheckAggrCheckParamBlob(self):
378	n/a	cur = self.con.cursor()
379	n/a	cur.execute("select checkType('blob', ?)", (memoryview(b"blob"),))
380	n/a	val = cur.fetchone()[0]
381	n/a	self.assertEqual(val, 1)
382	n/a
383	n/a	def CheckAggrCheckAggrSum(self):
384	n/a	cur = self.con.cursor()
385	n/a	cur.execute("delete from test")
386	n/a	cur.executemany("insert into test(i) values (?)", [(10,), (20,), (30,)])
387	n/a	cur.execute("select mysum(i) from test")
388	n/a	val = cur.fetchone()[0]
389	n/a	self.assertEqual(val, 60)
390	n/a
391	n/a	class AuthorizerTests(unittest.TestCase):
392	n/a	@staticmethod
393	n/a	def authorizer_cb(action, arg1, arg2, dbname, source):
394	n/a	if action != sqlite.SQLITE_SELECT:
395	n/a	return sqlite.SQLITE_DENY
396	n/a	if arg2 == 'c2' or arg1 == 't2':
397	n/a	return sqlite.SQLITE_DENY
398	n/a	return sqlite.SQLITE_OK
399	n/a
400	n/a	def setUp(self):
401	n/a	self.con = sqlite.connect(":memory:")
402	n/a	self.con.executescript("""
403	n/a	create table t1 (c1, c2);
404	n/a	create table t2 (c1, c2);
405	n/a	insert into t1 (c1, c2) values (1, 2);
406	n/a	insert into t2 (c1, c2) values (4, 5);
407	n/a	""")
408	n/a
409	n/a	# For our security test:
410	n/a	self.con.execute("select c2 from t2")
411	n/a
412	n/a	self.con.set_authorizer(self.authorizer_cb)
413	n/a
414	n/a	def tearDown(self):
415	n/a	pass
416	n/a
417	n/a	def test_table_access(self):
418	n/a	with self.assertRaises(sqlite.DatabaseError) as cm:
419	n/a	self.con.execute("select * from t2")
420	n/a	self.assertIn('prohibited', str(cm.exception))
421	n/a
422	n/a	def test_column_access(self):
423	n/a	with self.assertRaises(sqlite.DatabaseError) as cm:
424	n/a	self.con.execute("select c2 from t1")
425	n/a	self.assertIn('prohibited', str(cm.exception))
426	n/a
427	n/a	class AuthorizerRaiseExceptionTests(AuthorizerTests):
428	n/a	@staticmethod
429	n/a	def authorizer_cb(action, arg1, arg2, dbname, source):
430	n/a	if action != sqlite.SQLITE_SELECT:
431	n/a	raise ValueError
432	n/a	if arg2 == 'c2' or arg1 == 't2':
433	n/a	raise ValueError
434	n/a	return sqlite.SQLITE_OK
435	n/a
436	n/a	class AuthorizerIllegalTypeTests(AuthorizerTests):
437	n/a	@staticmethod
438	n/a	def authorizer_cb(action, arg1, arg2, dbname, source):
439	n/a	if action != sqlite.SQLITE_SELECT:
440	n/a	return 0.0
441	n/a	if arg2 == 'c2' or arg1 == 't2':
442	n/a	return 0.0
443	n/a	return sqlite.SQLITE_OK
444	n/a
445	n/a	class AuthorizerLargeIntegerTests(AuthorizerTests):
446	n/a	@staticmethod
447	n/a	def authorizer_cb(action, arg1, arg2, dbname, source):
448	n/a	if action != sqlite.SQLITE_SELECT:
449	n/a	return 2**32
450	n/a	if arg2 == 'c2' or arg1 == 't2':
451	n/a	return 2**32
452	n/a	return sqlite.SQLITE_OK
453	n/a
454	n/a
455	n/a	def suite():
456	n/a	function_suite = unittest.makeSuite(FunctionTests, "Check")
457	n/a	aggregate_suite = unittest.makeSuite(AggregateTests, "Check")
458	n/a	authorizer_suite = unittest.makeSuite(AuthorizerTests)
459	n/a	return unittest.TestSuite((
460	n/a	function_suite,
461	n/a	aggregate_suite,
462	n/a	authorizer_suite,
463	n/a	unittest.makeSuite(AuthorizerRaiseExceptionTests),
464	n/a	unittest.makeSuite(AuthorizerIllegalTypeTests),
465	n/a	unittest.makeSuite(AuthorizerLargeIntegerTests),
466	n/a	))
467	n/a
468	n/a	def test():
469	n/a	runner = unittest.TextTestRunner()
470	n/a	runner.run(suite())
471	n/a
472	n/a	if __name__ == "__main__":
473	n/a	test()